fix: Close the response body when a stream stops

build.gradle.kts

@@ -71,6 +71,7 @@ dependencies {
     api("com.squareup.okhttp3:okhttp:${Versions.okhttp}")
     testImplementation("org.mockito:mockito-core:1.10.19")
     testImplementation("com.launchdarkly:test-helpers:2.1.0")
+    testImplementation("com.squareup.okhttp3:mockwebserver:${Versions.okhttp}")
     testImplementation("com.google.guava:guava:33.5.0-jre")
     testImplementation("junit:junit:4.13.2")
     testImplementation("org.hamcrest:hamcrest-all:1.3")

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

@@ -97,6 +97,9 @@ public Result(InputStream inputStream, URI origin, Closeable closer, ResponseHea
 
       /**
        * The input stream that {@link EventSource} should read from.
+       * <p>
+       * When {@link EventSource} stops using the stream, it calls the {@link #getCloser() closer}
+       * and then closes this stream on the thread that reads from it.
        *
        * @return the input stream (must not be null)
        */

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

@@ -5,6 +5,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.io.InputStream;
 import java.net.URI;
 import java.net.URL;
 import java.util.ArrayList;
@@ -96,6 +97,7 @@ public class EventSource implements Closeable {
   // The following mutable fields are not volatile because they should only be
   // accessed from the thread that is reading from EventSource.
   private EventParser eventParser;
+  private InputStream eventParserInputStream;
   ErrorStrategy currentErrorStrategy;
   private long connectedTime;
   private long disconnectedTime;
@@ -337,6 +339,7 @@ private FaultEvent tryStart(boolean canReturnFaultEvent) throws StreamException
       logger.debug("Connected to SSE stream");
 
       ResponseHeaders headers = clientResult.getHeaders();
+      eventParserInputStream = clientResult.getInputStream();
       eventParser = new EventParser(
           clientResult.getInputStream(),
           clientResult.getOrigin(),
@@ -646,7 +649,7 @@ private StreamEvent requireEvent() throws StreamException {
       }
       disconnectedTime = System.currentTimeMillis();
       closeCurrentStream(false, false);
-      eventParser = null;
+      discardEventParser();
       if (applyErrorStrategy(e) == ErrorStrategy.Action.CONTINUE) {
         // At this point we're handling errors from reading the stream (not initial connection),
         // so we never have HTTP response headers available (headers is always null)
@@ -729,7 +732,6 @@ private boolean closeCurrentStream(boolean deliberatelyInterrupted, boolean shou
         }
       }
       if (oldReadingThread == Thread.currentThread()) {
-        eventParser = null;
         readyState.compareAndSet(ReadyState.OPEN, ReadyState.CLOSED);
         readyState.compareAndSet(ReadyState.CONNECTING, ReadyState.CLOSED);
         // If the current thread is not the reading thread, these fields will be updated the
@@ -738,8 +740,29 @@ private boolean closeCurrentStream(boolean deliberatelyInterrupted, boolean shou
 
       sleepNotifier.notify(); // in case we're sleeping in a reconnect delay, wake us up
     }
+    if (oldReadingThread == Thread.currentThread()) {
+      discardEventParser();
+    }
     return true;
   }
+
+  // Drops the parser for the current stream and closes its input stream. Call this only on
+  // the reading thread, because the input stream is not safe to close while another thread
+  // reads from it. Call it only after the connection closer has run. Closing the input
+  // stream releases the HTTP response. Cancelling the request does not release it, and an
+  // HTTP/2 connection stays in use until its response is released.
+  private void discardEventParser() {
+    eventParser = null;
+    InputStream inputStream = eventParserInputStream;
+    eventParserInputStream = null;
+    if (inputStream != null) {
+      try {
+        inputStream.close();
+      } catch (IOException e) {
+        logger.debug("Unexpected error when closing stream: {}", LogValues.exceptionSummary(e));
+      }
+    }
+  }
 
   /**
    * Builder for configuring {@link EventSource}.

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

@@ -0,0 +1,159 @@
+package com.launchdarkly.eventsource;
+
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.io.Closeable;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static com.launchdarkly.eventsource.MockConnectStrategy.ORIGIN;
+import static com.launchdarkly.eventsource.TestUtils.interruptOnAnotherThread;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.sameInstance;
+
+/**
+ * Tests that EventSource closes the input stream of a connection when it stops using
+ * the connection. It must close the stream on the reading thread, after the connection
+ * closer has run.
+ */
+@SuppressWarnings("javadoc")
+public class EventSourceStreamReleaseTest {
+  private static final String CLOSER_RAN = "closer ran";
+  private static final String STREAM_CLOSED = "input stream closed";
+  private static final String MESSAGE = "data: first\n\n";
+
+  @Rule public TestScopedLoggerRule testLogger = new TestScopedLoggerRule();
+
+  // Provides one stream that stays open until the connection closer runs, and records
+  // when the closer runs and when the input stream is closed.
+  private static class TrackedStreamHandler extends MockConnectStrategy.RequestHandler {
+    final List<String> actions = Collections.synchronizedList(new ArrayList<>());
+    volatile Thread streamClosingThread;
+    private final PipedOutputStream writeStream = new PipedOutputStream();
+    private final InputStream readStream;
+
+    TrackedStreamHandler(String data) throws IOException {
+      this(data, false);
+    }
+
+    TrackedStreamHandler(String data, boolean failOnClose) throws IOException {
+      readStream = new FilterInputStream(new PipedInputStream(writeStream)) {
+        @Override
+        public void close() throws IOException {
+          actions.add(STREAM_CLOSED);
+          streamClosingThread = Thread.currentThread();
+          super.close();
+          if (failOnClose) {
+            throw new IOException("deliberate error");
+          }
+        }
+      };
+      writeStream.write(data.getBytes());
+    }
+
+    @Override
+    protected InputStream getStream() {
+      return readStream;
+    }
+
+    @Override
+    protected Closeable makeCloser() {
+      return () -> {
+        actions.add(CLOSER_RAN);
+        writeStream.close();
+      };
+    }
+
+    void endStream() throws IOException {
+      writeStream.close();
+    }
+  }
+
+  private EventSource startEventSource(TrackedStreamHandler handler) throws Exception {
+    MockConnectStrategy mock = new MockConnectStrategy();
+    mock.configureRequests(handler, MockConnectStrategy.respondWithDataAndThenStayOpen(MESSAGE));
+    EventSource es = new EventSource.Builder(mock)
+        .errorStrategy(ErrorStrategy.alwaysContinue())
+        .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
+        .logger(testLogger.getLogger())
+        .build();
+    es.start();
+    assertThat(es.readAnyEvent(), equalTo(new MessageEvent("message", "first", null, ORIGIN)));
+    return es;
+  }
+
+  @Test
+  public void interruptOnReadingThreadClosesInputStreamAfterCloser() throws Exception {
+    TrackedStreamHandler handler = new TrackedStreamHandler(MESSAGE);
+    try (EventSource es = startEventSource(handler)) {
+      es.interrupt();
+
+      assertThat(handler.actions, contains(CLOSER_RAN, STREAM_CLOSED));
+      assertThat(handler.streamClosingThread, sameInstance(Thread.currentThread()));
+    }
+  }
+
+  @Test
+  public void stopOnReadingThreadClosesInputStreamAfterCloser() throws Exception {
+    TrackedStreamHandler handler = new TrackedStreamHandler(MESSAGE);
+    try (EventSource es = startEventSource(handler)) {
+      es.stop();
+
+      assertThat(handler.actions, contains(CLOSER_RAN, STREAM_CLOSED));
+    }
+  }
+
+  @Test
+  public void closeOnReadingThreadClosesInputStreamAfterCloser() throws Exception {
+    TrackedStreamHandler handler = new TrackedStreamHandler(MESSAGE);
+    EventSource es = startEventSource(handler);
+    es.close();
+
+    assertThat(handler.actions, contains(CLOSER_RAN, STREAM_CLOSED));
+  }
+
+  @Test
+  public void interruptFromAnotherThreadClosesInputStreamOnReadingThread() throws Exception {
+    TrackedStreamHandler handler = new TrackedStreamHandler(MESSAGE);
+    try (EventSource es = startEventSource(handler)) {
+      interruptOnAnotherThread(es);
+
+      assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByCallerException())));
+      assertThat(handler.actions, contains(CLOSER_RAN, STREAM_CLOSED));
+      assertThat(handler.streamClosingThread, sameInstance(Thread.currentThread()));
+    }
+  }
+
+  @Test
+  public void errorWhenClosingInputStreamDoesNotStopReconnect() throws Exception {
+    TrackedStreamHandler handler = new TrackedStreamHandler(MESSAGE, true);
+    try (EventSource es = startEventSource(handler)) {
+      es.interrupt();
+
+      assertThat(handler.actions, contains(CLOSER_RAN, STREAM_CLOSED));
+      assertThat(es.readAnyEvent(), equalTo(new StartedEvent()));
+      assertThat(es.readAnyEvent(), equalTo(new MessageEvent("message", "first", null, ORIGIN)));
+    }
+  }
+
+  @Test
+  public void endOfStreamClosesInputStream() throws Exception {
+    TrackedStreamHandler handler = new TrackedStreamHandler(MESSAGE);
+    try (EventSource es = startEventSource(handler)) {
+      handler.endStream();
+
+      assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException())));
+      assertThat(handler.actions, contains(CLOSER_RAN, STREAM_CLOSED));
+    }
+  }
+}

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

@@ -0,0 +1,123 @@
+package com.launchdarkly.eventsource;
+
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+import okhttp3.ConnectionPool;
+import okhttp3.Protocol;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.RecordedRequest;
+
+/**
+ * Tests of EventSource with real HTTP/2 connections. An HTTP/2 connection stays in use
+ * until its response is released, so these tests check that EventSource releases the
+ * response when it stops using a stream.
+ */
+@SuppressWarnings("javadoc")
+public class HttpConnectStrategyHttp2Test {
+  @Rule public TestScopedLoggerRule testLogger = new TestScopedLoggerRule();
+
+  // The body is larger than the client reads before the test stops the stream, so the
+  // stream is still active at that time.
+  private static MockResponse streamResponse(String eventData) {
+    StringBuilder body = new StringBuilder("data: " + eventData + "\n\n");
+    for (int i = 0; i < 50000; i++) {
+      body.append(":\n");
+    }
+    return new MockResponse()
+        .setHeader("Content-Type", "text/event-stream")
+        .setBody(body.toString());
+  }
+
+  private static MockWebServer startHttp2Server(MockResponse... responses) throws Exception {
+    MockWebServer server = new MockWebServer();
+    server.setProtocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE));
+    for (MockResponse response: responses) {
+      server.enqueue(response);
+    }
+    server.start();
+    return server;
+  }
+
+  private EventSource makeEventSource(MockWebServer server, ConnectionPool pool) {
+    return new EventSource.Builder(
+        ConnectStrategy.http(server.url("/").uri())
+          .clientBuilderActions(b -> b
+              .protocols(Collections.singletonList(Protocol.H2_PRIOR_KNOWLEDGE))
+              .connectionPool(pool)))
+        .errorStrategy(ErrorStrategy.alwaysContinue())
+        .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
+        .logger(testLogger.getLogger())
+        .build();
+  }
+
+  private static void readUntilMessage(EventSource es, String expectedData) throws Exception {
+    for (int i = 0; i < 10; i++) {
+      StreamEvent event = es.readAnyEvent();
+      if (event instanceof MessageEvent) {
+        assertThat(((MessageEvent)event).getData(), equalTo(expectedData));
+        return;
+      }
+    }
+    throw new AssertionError("did not receive message \"" + expectedData + "\"");
+  }
+
+  @Test
+  public void interruptOnReadingThreadReleasesConnection() throws Exception {
+    try (MockWebServer server = startHttp2Server(streamResponse("first"))) {
+      ConnectionPool pool = new ConnectionPool(5, 1, TimeUnit.MINUTES);
+      try (EventSource es = makeEventSource(server, pool)) {
+        readUntilMessage(es, "first");
+        assertThat(pool.idleConnectionCount(), equalTo(0));
+
+        es.interrupt();
+
+        assertThat(pool.connectionCount(), equalTo(1));
+        assertThat(pool.idleConnectionCount(), equalTo(1));
+      }
+    }
+  }
+
+  @Test
+  public void reconnectAfterInterruptOnReadingThreadUsesNewConnectionIfPoolKeepsNoIdleConnections()
+      throws Exception {
+    try (MockWebServer server = startHttp2Server(streamResponse("first"), streamResponse("second"))) {
+      ConnectionPool pool = new ConnectionPool(0, 1, TimeUnit.SECONDS);
+      try (EventSource es = makeEventSource(server, pool)) {
+        readUntilMessage(es, "first");
+
+        es.interrupt();
+        assertThat(pool.connectionCount(), equalTo(0));
+
+        readUntilMessage(es, "second");
+
+        RecordedRequest request1 = server.takeRequest(1, TimeUnit.SECONDS);
+        RecordedRequest request2 = server.takeRequest(1, TimeUnit.SECONDS);
+        assertThat(request1.getSequenceNumber(), equalTo(0));
+        // A sequence number of 0 means that the request was the first one on its connection.
+        assertThat(request2.getSequenceNumber(), equalTo(0));
+      }
+    }
+  }
+
+  @Test
+  public void stopOnReadingThreadReleasesConnection() throws Exception {
+    try (MockWebServer server = startHttp2Server(streamResponse("first"))) {
+      ConnectionPool pool = new ConnectionPool(5, 1, TimeUnit.MINUTES);
+      try (EventSource es = makeEventSource(server, pool)) {
+        readUntilMessage(es, "first");
+
+        es.stop();
+
+        assertThat(pool.idleConnectionCount(), equalTo(1));
+      }
+    }
+  }
+}