You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cxf.apache.org by re...@apache.org on 2016/09/10 14:41:21 UTC
[13/37] cxf git commit: [CXF-6618] Resuming the async responce only
when the batch is started
[CXF-6618] Resuming the async responce only when the batch is started
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/76edf935
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/76edf935
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/76edf935
Branch: refs/heads/master-jaxrs-2.1
Commit: 76edf9350e7640860b66196ad40396d54bbcb2b2
Parents: 46438c4
Author: Sergey Beryozkin <sb...@gmail.com>
Authored: Tue Sep 6 16:37:07 2016 +0100
Committer: Sergey Beryozkin <sb...@gmail.com>
Committed: Tue Sep 6 16:37:07 2016 +0100
----------------------------------------------------------------------
.../main/java/demo/jaxrs/server/SparkJob.java | 36 ++++++++++++++++++
.../jaxrs/server/SparkStreamingListener.java | 40 +++++++++++++++++++-
.../demo/jaxrs/server/StreamingService.java | 12 +++---
3 files changed, 79 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/76edf935/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/SparkJob.java
----------------------------------------------------------------------
diff --git a/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/SparkJob.java b/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/SparkJob.java
new file mode 100644
index 0000000..010ddf3
--- /dev/null
+++ b/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/SparkJob.java
@@ -0,0 +1,36 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package demo.jaxrs.server;
+
+import javax.ws.rs.container.AsyncResponse;
+
+public class SparkJob implements Runnable {
+ private AsyncResponse ac;
+ private SparkStreamingListener sparkListener;
+ public SparkJob(AsyncResponse ac, SparkStreamingListener sparkListener) {
+ this.ac = ac;
+ this.sparkListener = sparkListener;
+ }
+ @Override
+ public void run() {
+ sparkListener.waitForBatchStarted();
+ ac.resume(sparkListener.getStreamOut());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/cxf/blob/76edf935/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/SparkStreamingListener.java
----------------------------------------------------------------------
diff --git a/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/SparkStreamingListener.java b/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/SparkStreamingListener.java
index 3fd7284..8a891c2 100644
--- a/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/SparkStreamingListener.java
+++ b/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/SparkStreamingListener.java
@@ -1,3 +1,21 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package demo.jaxrs.server;
import org.apache.spark.streaming.scheduler.StreamingListener;
@@ -12,7 +30,8 @@ import org.apache.spark.streaming.scheduler.StreamingListenerReceiverStopped;
public class SparkStreamingListener implements StreamingListener {
private SparkStreamingOutput streamOutput;
-
+ private boolean batchStarted;
+
public SparkStreamingListener(SparkStreamingOutput streamOutput) {
this.streamOutput = streamOutput;
}
@@ -23,7 +42,9 @@ public class SparkStreamingListener implements StreamingListener {
}
@Override
- public void onBatchStarted(StreamingListenerBatchStarted event) {
+ public synchronized void onBatchStarted(StreamingListenerBatchStarted event) {
+ batchStarted = true;
+ notify();
}
@Override
@@ -49,5 +70,20 @@ public class SparkStreamingListener implements StreamingListener {
@Override
public void onReceiverStopped(StreamingListenerReceiverStopped arg0) {
}
+
+ public SparkStreamingOutput getStreamOut() {
+ return streamOutput;
+ }
+
+ public synchronized void waitForBatchStarted() {
+ while (!batchStarted) {
+ try {
+ this.wait();
+ } catch (InterruptedException ex) {
+ // continue
+ }
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/76edf935/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/StreamingService.java
----------------------------------------------------------------------
diff --git a/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/StreamingService.java b/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/StreamingService.java
index f986225..0e94b3c 100644
--- a/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/StreamingService.java
+++ b/distribution/src/main/release/samples/jax_rs/spark/src/main/java/demo/jaxrs/server/StreamingService.java
@@ -66,19 +66,17 @@ public class StreamingService {
@Produces("text/plain")
public void getStream(@Suspended AsyncResponse async, InputStream is) {
try {
+ SparkStreamingOutput streamOut = new SparkStreamingOutput(jssc);
+ SparkStreamingListener sparkListener = new SparkStreamingListener(streamOut);
+ jssc.addStreamingListener(sparkListener);
+
JavaReceiverInputDStream<String> receiverStream =
jssc.receiverStream(new InputStreamReceiver(is));
- SparkStreamingOutput streamOut = new SparkStreamingOutput(jssc);
- jssc.addStreamingListener(new SparkStreamingListener(streamOut));
JavaPairDStream<String, Integer> wordCounts = createOutputDStream(receiverStream);
wordCounts.foreachRDD(new OutputFunction(streamOut));
jssc.start();
- executor.execute(new Runnable() {
- public void run() {
- async.resume(streamOut);
- }
- });
+ executor.execute(new SparkJob(async, sparkListener));
} catch (Exception ex) {
// the compiler does not allow to catch SparkException directly
if (ex instanceof SparkException) {