in streaming data mode, throw an exception if stream is closed during…

… an event (#68)

* in streaming data mode, throw an exception if stream is closed during an event

* one more unit test

src/main/java/com/launchdarkly/eventsource/EventParser.java

@@ -424,7 +424,10 @@ public int read(byte[] b, int off, int len) throws IOException {
         // (B) We must ask lineParser to give us another chunk of a not-yet-finished line.
         if (!lineEnded) {
           if (!canGetNextChunk()) {
-            return -1; // EOF
+            // The underlying SSE stream has run out of data while we were still trying to
+            // read the rest of the message. This is an abnormal condition, so we'll treat
+            // it as an exception, rather than just returning -1 to indicate EOF.
+            throw new StreamClosedWithIncompleteMessageException();
           }
           haveChunk = true;
           continue; // We'll go to (A) in the next loop
@@ -433,10 +436,15 @@ public int read(byte[] b, int off, int len) throws IOException {
         // (C) The previous line was done; ask lineParser to give us the next line (or at
         // least the first chunk of it).
         if (!canGetNextChunk()) {
-          return -1; // EOF
+          // See comment above about abnormal termination. Even if we just finished
+          // reading a complete line of data, the message is incomplete because we didn't
+          // see a blank line.
+          throw new StreamClosedWithIncompleteMessageException();
         }
         if (lineEnded && chunkSize == 0) {
-          // Blank line means end of message - close this stream and return EOF.
+          // Blank line means end of message - close this stream and return EOF. This is a
+          // normal condition: the stream of data for this message is done because the
+          // message is finished.
           closed.set(true);
           resetState();
           return -1;

src/main/java/com/launchdarkly/eventsource/EventSource.java

@@ -962,9 +962,10 @@ public Builder logger(LDLogger logger) {
      * If you set it to {@code true}, it will instead call the handler as soon as it sees a {@code data} field--
      * setting {@link MessageEvent#getDataReader()} to a {@link java.io.Reader} that reads directly from the data as
      * it arrives. The EventSource will perform any necessary parsing under the covers, so that for instance if there
-     * are multiple {@code data:} lines in the event, the {@link java.io.Reader} will emit a newline character between
-     * each and will not see the "data:" field names. The {@link java.io.Reader} will report "end of stream" as soon
-     * as the event is terminated normally by a blank line.
+     * are multiple {@code data:} lines in the event, the Reader will emit a newline character between
+     * each and will not see the "data:" field names. The Reader will report "end of stream" as soon
+     * as the event is terminated normally by a blank line. If the stream is closed before normal termination of
+     * the event, the Reader will throw a {@link StreamClosedWithIncompleteMessageException}.
      * <p>
      * This mode is designed for applications that expect very large data items to be delivered over SSE. Use it
      * with caution, since there are several limitations:
@@ -978,7 +979,8 @@ public Builder logger(LDLogger logger) {
      * <li> The SSE protocol specifies that an event should be processed only if it is terminated by a blank line, but
      * in this mode the handler will receive the event as soon as a {@code data:} field appears-- so, if the stream
      * happens to cut off abnormally without a trailing blank line, technically you will be receiving an incomplete
-     * event that should have been ignored. </li>
+     * event that should have been ignored. You will know this has happened ifbecause reading from the Reader throws
+     * a {@link StreamClosedWithIncompleteMessageException}.</li>
      * </ul>  
      * 
      * @param streamEventData true if events should be dispatched immediately with asynchronous data rather than

src/main/java/com/launchdarkly/eventsource/MessageEvent.java

@@ -195,7 +195,13 @@ public String getData() {
    * <p>
    * The method will never return {@code null}; every event has data, even if the data is empty
    * (zero length).
-   *  
+   * <p>
+   * If the stream connection is closed before a complete SSE message has been received (that is,
+   * before the usual blank line that would terminate a message), then instead of a normal EOF,
+   * the Reader will throw a {@link StreamClosedWithIncompleteMessageException}. If this happens,
+   * the application should generally discard the entire {@link MessageEvent} and not try to
+   * process it further. 
+   *
    * @return a reader for the event data
    * @since 2.6.0
    */

src/main/java/com/launchdarkly/eventsource/StreamClosedWithIncompleteMessageException.java

@@ -0,0 +1,12 @@
+package com.launchdarkly.eventsource;
+
+import java.io.IOException;
+
+/**
+ * @author elibishop
+ *
+ */
+@SuppressWarnings("serial")
+public class StreamClosedWithIncompleteMessageException extends IOException {
+
+}

src/test/java/com/launchdarkly/eventsource/EventParserStreamingDataTest.java

@@ -1,5 +1,6 @@
 package com.launchdarkly.eventsource;
 
+import org.junit.Assert;
 import org.junit.Test;
 
 import java.io.IOException;
@@ -10,6 +11,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.fail;
 
 @SuppressWarnings("javadoc")
 public class EventParserStreamingDataTest extends EventParserBaseTest {
@@ -179,7 +181,22 @@ public void incompletelyReadEventIsSkippedIfAnotherMessageIsRead() throws Except
   }
 
   @Test
-  public void streamIsClosedBeforeEndOfEvent() throws Exception {
+  public void streamIsClosedImmediatelyAfterEndOfEvent() throws Exception {
+    String streamData = "data: hello\n\n";
+    initParser(100, true);
+    processData(streamData);
+
+    MessageEvent e1 = awaitMessageEvent();
+    assertThat(e1.isStreamingData(), is(true));
+    assertThat(readUpToLimit(e1.getDataReader(), 5), equalTo("hello"));
+
+    closeStream();
+
+    assertThat(e1.getDataReader().read(), equalTo(-1)); // normal EOF, not an error
+  }
+
+  @Test
+  public void streamIsClosedBeforeEndOfEventAtEndOfLine() throws Exception {
     String streamData = "data: hello\n";
     initParser(100, true);
     processData(streamData);
@@ -190,7 +207,28 @@ public void streamIsClosedBeforeEndOfEvent() throws Exception {
 
     closeStream();
 
-    assertThat(e1.getDataReader().read(), equalTo(-1));
+    try {
+      e1.getDataReader().read();
+      fail("expected exception");
+    } catch (StreamClosedWithIncompleteMessageException e) {}
+  }
+
+  @Test
+  public void streamIsClosedBeforeEndOfEventWithinLine() throws Exception {
+    String streamData = "data: hello";
+    initParser(100, true);
+    processData(streamData);
+
+    MessageEvent e1 = awaitMessageEvent();
+    assertThat(e1.isStreamingData(), is(true));
+    assertThat(readUpToLimit(e1.getDataReader(), 5), equalTo("hello"));
+
+    closeStream();
+
+    try {
+      e1.getDataReader().read();
+      fail("expected exception");
+    } catch (StreamClosedWithIncompleteMessageException e) {}
   }
 
   @Test