You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by vr...@apache.org on 2015/09/24 04:37:37 UTC
[18/50] [abbrv] incubator-apex-core git commit: APEX-121 #resolve
APEX-121 #resolve
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/89aefea4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/89aefea4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/89aefea4
Branch: refs/heads/feature-module
Commit: 89aefea4c0b5ac8b40efd2a191b81142a8c302f9
Parents: c349090
Author: Gaurav <ga...@datatorrent.com>
Authored: Sat Sep 12 22:50:11 2015 -0700
Committer: Gaurav <ga...@datatorrent.com>
Committed: Sat Sep 12 22:50:11 2015 -0700
----------------------------------------------------------------------
.../com/datatorrent/stram/stream/BufferServerPublisher.java | 9 ++++++++-
1 file changed, 8 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/89aefea4/engine/src/main/java/com/datatorrent/stram/stream/BufferServerPublisher.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/stream/BufferServerPublisher.java b/engine/src/main/java/com/datatorrent/stram/stream/BufferServerPublisher.java
index d8e09c8..39cf667 100644
--- a/engine/src/main/java/com/datatorrent/stram/stream/BufferServerPublisher.java
+++ b/engine/src/main/java/com/datatorrent/stram/stream/BufferServerPublisher.java
@@ -112,7 +112,14 @@ public class BufferServerPublisher extends Publisher implements ByteCounterStrea
* if there is any state write that for the subscriber before we write the data.
*/
if (dsp.state != null) {
- write(DataTuple.getSerializedTuple(MessageType.CODEC_STATE_VALUE, dsp.state));
+ array = DataTuple.getSerializedTuple(MessageType.CODEC_STATE_VALUE, dsp.state);
+ try {
+ while (!write(array)) {
+ sleep(5);
+ }
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ }
}
/*
* Now that the state if any has been sent, we can proceed with the actual data we want to send.