launchdarkly
diff --git a/‎build.gradle.kts
+12-10 b/‎build.gradle.kts
+12-10
diff --git a/‎contract-tests/service/src/main/java/ssetest/StreamEntity.java
+46-32 b/‎contract-tests/service/src/main/java/ssetest/StreamEntity.java
+46-32
diff --git a/‎gradle.properties
+1-1 b/‎gradle.properties
+1-1
diff --git a/‎src/main/java/com/launchdarkly/eventsource/AsyncEventHandler.java
-126 b/‎src/main/java/com/launchdarkly/eventsource/AsyncEventHandler.java
-126
diff --git a/‎src/main/java/com/launchdarkly/eventsource/CommentEvent.java
+53 b/‎src/main/java/com/launchdarkly/eventsource/CommentEvent.java
+53
@@ -69,7 +69,7 @@ dependencies {
     api("com.launchdarkly:launchdarkly-logging:${Versions.launchdarklyLogging}")
     api("com.squareup.okhttp3:okhttp:${Versions.okhttp}")
     testImplementation("org.mockito:mockito-core:1.10.19")
-    testImplementation("com.launchdarkly:test-helpers:1.0.0")
+    testImplementation("com.launchdarkly:test-helpers:2.0.1")
     testImplementation("com.google.guava:guava:30.1-jre")
     testImplementation("junit:junit:4.12")
     testImplementation("org.hamcrest:hamcrest-all:1.3")
@@ -122,24 +122,26 @@ tasks.jacocoTestCoverageVerification.configure {
     violationRules {
         val knownMissedLinesForMethods = mapOf(
             // The key for each of these items is the complete method signature minus the "com.launchdarkly.eventsource." prefix.
-            "AsyncEventHandler.acquire()" to 2,
-            "AsyncEventHandler.execute(java.lang.Runnable)" to 3,
+            "EventParser.IncrementalMessageDataInputStream.read()" to 7,
+            "EventParser.IncrementalMessageDataInputStream.read(byte[])" to 1,
+            "EventParser.IncrementalMessageDataInputStream.read(byte[], int, int)" to 2,
+            "EventParser.IncrementalMessageDataInputStream.canGetNextChunk()" to 3,
             "EventSource.awaitClosed(long, java.util.concurrent.TimeUnit)" to 2,
             "EventSource.awaitClosed(java.time.Duration)" to 1,
-            "EventSource.handleSuccessfulResponse(okhttp3.Response)" to 2,
-            "EventSource.maybeReconnectDelay(int, long)" to 2,
-            "EventSource.run()" to 3,
-            "EventSource.Builder.createInitialClientBuilder()" to 1,
-            "EventSource.Builder.defaultTrustManager()" to 2,
+            "Helpers.utf8ByteArrayOutputStreamToString(java.io.ByteArrayOutputStream)" to 2,
+            "HttpConnectStrategy.defaultTrustManager()" to 2,
+            "HttpConnectStrategy.Client.awaitClosed(long)" to 1,
+            "HttpConnectStrategy.Client.createHttpClient()" to 2,
             "MessageEvent.getData()" to 2,
             "ModernTLSSocketFactory.createSocket(java.lang.String, int)" to 1,
             "ModernTLSSocketFactory.createSocket(java.lang.String, int, java.net.InetAddress, int)" to 1,
             "ModernTLSSocketFactory.createSocket(java.net.InetAddress, int)" to 1,
             "ModernTLSSocketFactory.createSocket(java.net.InetAddress, int, java.net.InetAddress, int)" to 1,
             "ModernTLSSocketFactory.createSocket(java.net.Socket, java.lang.String, int, boolean)" to 1,
             "ModernTLSSocketFactory.getDefaultCipherSuites()" to 1,
-            "ModernTLSSocketFactory.getSupportedCipherSuites()" to 1
-        )
+            "ModernTLSSocketFactory.getSupportedCipherSuites()" to 1,
+            "background.BackgroundEventSource.dispatchEvent(com.launchdarkly.eventsource.StreamEvent)" to 3
+            )
 
         knownMissedLinesForMethods.forEach { (signature, maxMissedLines) ->
             if (maxMissedLines > 0) {  // < 0 means skip entire method
 
@@ -8,10 +8,10 @@
 import okhttp3.*;
 import ssetest.Representations.*;
 
-public class StreamEntity implements EventHandler {
+public class StreamEntity {
   private final TestService owner;
   private final String id;
-  private final EventSource stream;
+  private final EventSource eventSource;
   private final StreamOptions options;
   private final AtomicInteger callbackMessageCounter = new AtomicInteger(0);
   private final LDLogger logger;
@@ -25,74 +25,88 @@ public StreamEntity(TestService owner, String id, StreamOptions options, LDLogAd
     this.logger = LDLogger.withAdapter(logAdapter, options.tag);
     logger.info("Opening stream to {}", options.streamUrl);
 
-    EventSource.Builder eb = new EventSource.Builder(this, URI.create(options.streamUrl))
-      .logger(logger.subLogger("stream"));
+    HttpConnectStrategy connectStrategy = ConnectStrategy.http(URI.create(options.streamUrl));
+    if (options.readTimeoutMs != null) {
+      connectStrategy = connectStrategy.readTimeout((long)options.readTimeoutMs, null);
+    }
+    if (options.method != null) {
+      connectStrategy = connectStrategy.methodAndBody(options.method, 
+        options.body == null ? null :
+          RequestBody.create(options.body,
+            MediaType.parse(options.headers.get("content-type") == null ?
+              "text/plain; charset=utf-8" : options.headers.get("content-type")))
+      );
+    }
     if (options.headers != null) {
-      Headers.Builder hb = new Headers.Builder();
       for (String name: options.headers.keySet()) {
-        hb.add(name, options.headers.get(name));
+        connectStrategy = connectStrategy.header(name, options.headers.get(name));
       }
-      eb.headers(hb.build());
     }
+
+    EventSource.Builder eb = new EventSource.Builder(connectStrategy)
+      .errorStrategy(ErrorStrategy.alwaysContinue())
+      .logger(logger.subLogger("stream"));
     if (options.initialDelayMs != null) {
-      eb.reconnectTime(options.initialDelayMs, null);
-    }
-    if (options.readTimeoutMs != null) {
-      eb.readTimeout(options.readTimeoutMs, null);
+      eb.retryDelay((long)options.initialDelayMs, null);
     }
     if (options.lastEventId != null) {
       eb.lastEventId(options.lastEventId);
     }
-    if (options.method != null) {
-      eb.method(options.method);
-    }
-    if (options.body != null) {
-      String contentType = options.headers == null ? null : options.headers.get("content-type");
-      eb.body(RequestBody.create(options.body,
-        MediaType.parse(contentType == null ? "text/plain; charset=utf-8" : contentType)));
-    }
-    this.stream = eb.build();
-    
-    this.stream.start();
+
+    this.eventSource = eb.build();
+    new Thread(() -> {
+      for (StreamEvent event: this.eventSource.anyEvents()) {
+        handleEvent(event);
+      }
+    }).start();
   }
 
   public boolean doCommand(String command) {
     logger.info("Test harness sent command: {}", command);
     if (command.equals("restart")) {
-      stream.restart();
+      eventSource.interrupt();
       return true;
     }
     return false;
   }
 
   public void close() {
     closed = true;
-    stream.close();
+    eventSource.close();
     owner.forgetStream(id);
     logger.info("Test ended");
   }
 
-  public void onOpen() {}
-  
-  public void onClosed() {}
+  private void handleEvent(StreamEvent event) {
+    if (event instanceof MessageEvent) {
+      onMessage((MessageEvent)event);
+    } else if (event instanceof CommentEvent) {
+      onComment(((CommentEvent)event).getText());
+    } else if (event instanceof FaultEvent) {
+      onError(((FaultEvent)event).getCause());
+    }
+  }
 
-  public void onMessage(String name, MessageEvent e) {
-    logger.info("Received event from stream ({})", name);
+  private void onMessage(MessageEvent e) {
+    logger.info("Received event from stream ({})", e.getEventName());
     Message m = new Message("event");
     m.event = new EventMessage();
-    m.event.type = name;
+    m.event.type = e.getEventName();
     m.event.data = e.getData();
     m.event.id = e.getLastEventId();
     writeMessage(m);
   }
 
-  public void onComment(String comment) {
+  private void onComment(String comment) {
     Message m = new Message("comment");
     m.comment = comment;
     writeMessage(m);
   }
 
-  public void onError(Throwable t) {
+  private void onError(Throwable t) {
+    if (t instanceof StreamClosedByCallerException) {
+      return; // the SSE contract tests don't want to see this non-error
+    }
     logger.info("Received error from stream: {}", t.toString());
     Message m = new Message("error");
     m.error = t.toString();
 
@@ -1,4 +1,4 @@
-version=3.0.0
+version=4.0.0-SNAPSHOT
 ossrhUsername=
 ossrhPassword=
 
 
@@ -0,0 +1,53 @@
+package com.launchdarkly.eventsource;
+
+import java.util.Objects;
+
+/**
+ * Describes a comment line received from the stream.
+ * <p>
+ * An SSE comment is a line that starts with a colon. There is no defined meaning for this
+ * in the SSE specification, and most clients ignore it. It may be used to provide a
+ * periodic heartbeat from the server to keep connections from timing out.
+ * 
+ * @since 4.0.0
+ */
+public final class CommentEvent implements StreamEvent {
+  private final String text;
+  
+  /**
+   * Creates an instance.
+   *
+   * @param text the comment text, not including the leading colon
+   */
+  public CommentEvent(String text) {
+    this.text = text;
+  }
+
+  /**
+   * Returns the comment text, not including the leading colon.
+   *
+   * @return the text
+   */
+  public String getText() {
+    return text;
+  }
+  
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+
+    CommentEvent that = (CommentEvent) o;
+    return Objects.equals(text, that.text);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(text);
+  }
+  
+  @Override
+  public String toString() {
+    return "CommentEvent(" + text + ")";
+  }
+}
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-version=3.0.0`
	`1`	`+version=4.0.0-SNAPSHOT`
`2`	`2`	`ossrhUsername=`
`3`	`3`	`ossrhPassword=`
`4`	`4`