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..595f29e --- /dev/null +++ b/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 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)); + } + } +} 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)); + } + } + } +}