From be68142dc3faaf8196f5d01b310db6914a6e1e7e Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:20:14 -0700 Subject: [PATCH 1/2] fix: Close the response body when a stream stops EventSource stopped a stream by cancelling the request, but it did not close the response body when the stream stopped on the reading thread, for example when interrupt() is called inside the anyEvents() loop. On HTTP/2, cancelling only resets the stream, so the request stays attached to the connection and the connection is never released. The reading thread now closes the input stream whenever it stops using a stream, after the connection closer has run. --- build.gradle.kts | 1 + .../eventsource/ConnectStrategy.java | 3 + .../launchdarkly/eventsource/EventSource.java | 27 +++- .../EventSourceStreamReleaseTest.java | 140 ++++++++++++++++++ .../HttpConnectStrategyHttp2Test.java | 123 +++++++++++++++ 5 files changed, 292 insertions(+), 2 deletions(-) create mode 100644 src/test/java/com/launchdarkly/eventsource/EventSourceStreamReleaseTest.java create mode 100644 src/test/java/com/launchdarkly/eventsource/HttpConnectStrategyHttp2Test.java diff --git a/build.gradle.kts b/build.gradle.kts index e581a15..5de278a 100644 --- a/build.gradle.kts +++ b/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") diff --git a/src/main/java/com/launchdarkly/eventsource/ConnectStrategy.java b/src/main/java/com/launchdarkly/eventsource/ConnectStrategy.java index fe7f30c..591fa47 100644 --- a/src/main/java/com/launchdarkly/eventsource/ConnectStrategy.java +++ b/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. + *

+ * 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) */ diff --git a/src/main/java/com/launchdarkly/eventsource/EventSource.java b/src/main/java/com/launchdarkly/eventsource/EventSource.java index 674572f..8acf9ad 100644 --- a/src/main/java/com/launchdarkly/eventsource/EventSource.java +++ b/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}. diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceStreamReleaseTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceStreamReleaseTest.java new file mode 100644 index 0000000..3a15939 --- /dev/null +++ b/src/test/java/com/launchdarkly/eventsource/EventSourceStreamReleaseTest.java @@ -0,0 +1,140 @@ +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 actions = Collections.synchronizedList(new ArrayList<>()); + volatile Thread streamClosingThread; + private final PipedOutputStream writeStream = new PipedOutputStream(); + private final InputStream readStream; + + TrackedStreamHandler(String data) throws IOException { + readStream = new FilterInputStream(new PipedInputStream(writeStream)) { + @Override + public void close() throws IOException { + actions.add(STREAM_CLOSED); + streamClosingThread = Thread.currentThread(); + super.close(); + } + }; + 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 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)); + } + } +} diff --git a/src/test/java/com/launchdarkly/eventsource/HttpConnectStrategyHttp2Test.java b/src/test/java/com/launchdarkly/eventsource/HttpConnectStrategyHttp2Test.java new file mode 100644 index 0000000..6462e47 --- /dev/null +++ b/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)); + } + } + } +} From 3114c808dd59cab4c26cad778c44f2a6b1fcfdfe Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:23:44 -0700 Subject: [PATCH 2/2] test: Cover an error when closing the stream input --- .../EventSourceStreamReleaseTest.java | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceStreamReleaseTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceStreamReleaseTest.java index 3a15939..595f29e 100644 --- a/src/test/java/com/launchdarkly/eventsource/EventSourceStreamReleaseTest.java +++ b/src/test/java/com/launchdarkly/eventsource/EventSourceStreamReleaseTest.java @@ -43,12 +43,19 @@ private static class TrackedStreamHandler extends MockConnectStrategy.RequestHan 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()); @@ -127,6 +134,18 @@ public void interruptFromAnotherThreadClosesInputStreamOnReadingThread() throws } } + @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);