Application does not exit when cancelling a SSE connection

[bashopman]

This bug report results from this question on Stack Overflow.

Problem:

I'm trying to create a client that can receive events from a server using SSE (server-sent events) and I'm using OkHttp3 to do that, combined with the OkHttp3-sse library (com.squareup.okhttp3:okhttp:4.1.0 & com.squareup.okhttp3:okhttp-sse:4.1.0). I am able to connect to the server and receive the events just fine, but at some point in time, I want to disconnect the client and shutdown my application. That is were the problem is. The connection gets closed, but the application is not shutting down, probably due to the ConnectionPool not closing.

Attempted:

As there is hardly any documentation on the use of the OkHttp3-sse library, I have tried to reverse engineer from the code what to do. I have tried to:

• call RealEventSource.cancel() which is disconnecting, but is hanging on a thread
• call OkHttpClient.dispatcher().cancelAll() which is also disconnecting, but remains hanging on a thread
• call Response.close() which is passed to EventSourceListener.onOpen()

Finding:

As an alternative, I have looked at https://github.com/heremaps/oksse which has the RealServerSentEvent.close() call which is doing what I expect. It closes the connection and stops all threads allowing the application to shutdown completely. I have looked at the implementation of the RealServerSentEvent class to see how the implementation differs from RealEventSource. I think the difference is in the conditions of the read loop:

RealEventSource line https://github.com/square/okhttp/blob/master/okhttp-sse/src/main/java/okhttp3/internal/sse/RealEventSource.kt#L75:

      try {
        listener.onOpen(this, response)
        while (reader.processNextEvent()) {
        }
      } catch (e: Exception) {
        listener.onFailure(this, e, response)
        return
      }

compares to RealServerSentEvent line https://github.com/heremaps/oksse/blob/master/src/main/java/com/here/oksse/RealServerSentEvent.java#L94:

            listener.onOpen(this, response);

            //noinspection StatementWithEmptyBody
            while (call != null && !call.isCanceled() && sseReader.read()) {
            }

The OkSSE implementation includes the condition if the call.isCanceled() in the loop, where as the OkHttp3-sse does not. I suspect this is causing OkHttp3-sse not to exit, but I might be mistaken.

Test:

Code:

package okhttp3.sse;

import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.junit.Test;

import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

public class SseDisconnectTest {

    private static final Logger LOGGER = Logger.getLogger(SseDisconnectTest.class.getSimpleName());
    private static final String URL = "https://proxy.streamdata.io/https://api.myjson.com/bins/jixid";

    public static void main(String[] args) {
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                super.run();
                LOGGER.info("exiting application");
            }
        });
        OkHttpClient client = new OkHttpClient.Builder()
                .readTimeout(0, TimeUnit.SECONDS)
                .build();
        EventSourceListener listener = new Listener();
        Request request = new Request.Builder()
                .url(URL)
                .header("Accept-Encoding", "")
                .header("Accept", "text/event-stream")
                .header("Cache-Control", "no-cache")
                .build();
        final EventSource eventSource = EventSources.createFactory(client).newEventSource(request, listener);
        try {
            Thread.sleep(10_000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        eventSource.cancel();
        LOGGER.info("cancelled");
    }

    static class Listener extends EventSourceListener {
        @Override
        public void onOpen(@NotNull EventSource eventSource, @NotNull Response response) {
            super.onOpen(eventSource, response);
            LOGGER.info("onOpen");
        }

        @Override
        public void onEvent(@NotNull EventSource eventSource, @Nullable String id, @Nullable String type, @NotNull String data) {
            super.onEvent(eventSource, id, type, data);
            LOGGER.info(data);
        }

        @Override
        public void onClosed(@NotNull EventSource eventSource) {
            super.onClosed(eventSource);
            LOGGER.info("closed");
        }

        @Override
        public void onFailure(@NotNull EventSource eventSource, @Nullable Throwable t, @Nullable Response response) {
            super.onFailure(eventSource, t, response);
            t.printStackTrace();
        }
    }
}

Resulting log:

Oct 09, 2019 5:42:31 PM okhttp3.sse.SseDisconnectTest$Listener onOpen
INFO: onOpen
Oct 09, 2019 5:42:31 PM okhttp3.sse.SseDisconnectTest$Listener onEvent
INFO: {"name":"oksse","test":1}
Oct 09, 2019 5:42:40 PM okhttp3.sse.SseDisconnectTest main
INFO: cancelled
java.net.SocketException: Socket closed
    at java.net.SocketInputStream.socketRead0(Native Method)
    at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
    at java.net.SocketInputStream.read(SocketInputStream.java:171)
    at java.net.SocketInputStream.read(SocketInputStream.java:141)
    at sun.security.ssl.InputRecord.readFully(InputRecord.java:465)
    at sun.security.ssl.InputRecord.read(InputRecord.java:503)
    at sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:975)
    at sun.security.ssl.SSLSocketImpl.readDataRecord(SSLSocketImpl.java:933)
    at sun.security.ssl.AppInputStream.read(AppInputStream.java:105)
    at okio.InputStreamSource.read(Okio.kt:102)
    at okio.AsyncTimeout$source$1.read(AsyncTimeout.kt:159)
    at okio.RealBufferedSource.read(RealBufferedSource.kt:41)
    at okhttp3.internal.http1.Http1ExchangeCodec$AbstractSource.read(Http1ExchangeCodec.kt:352)
    at okhttp3.internal.http1.Http1ExchangeCodec$UnknownLengthSource.read(Http1ExchangeCodec.kt:488)
    at okhttp3.internal.connection.Exchange$ResponseBodySource.read(Exchange.kt:279)
    at okio.RealBufferedSource.select(RealBufferedSource.kt:93)
    at okhttp3.internal.sse.ServerSentEventReader.processNextEvent(ServerSentEventReader.kt:50)
    at okhttp3.internal.sse.RealEventSource.processResponse(RealEventSource.kt:75)
    at okhttp3.internal.sse.RealEventSource.onResponse(RealEventSource.kt:46)
    at okhttp3.RealCall$AsyncCall.run(RealCall.kt:138)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Oct 09, 2019 5:43:40 PM okhttp3.sse.SseDisconnectTest$1 run
INFO: exiting application

Note that the connection is cancelled at 5:42:40PM, while the application exits at 5:43:40PM after some timeout.

[bashopman]

Seems related to #5542 .

[bashopman]

Is it possible to provide a timeline to get this fixed? The answer on Stack Overflow suggested it could be done on short notice. Thanks.

[JakeWharton]

This project is maintained by volunteer effort. If you need a fix on some timeframe it's best to pursue one yourself and contribute a PR.

[swankjesse]

Try shutting down the dispatcher's executor service?

[bashopman]

I'm sorry. Didn't mean to be rude. I guess I misunderstood the answer on Stack Overflow. I'll try to create a fix and provide a pull request taking swankjesse's suggestion into account.

[bashopman]

Using the before mentioned versions of OkHttp and OkHttp-sse and swankjesses suggestion regarding shutting down the executor service, I was able to properly shut down the application by calling:

eventSource.cancel();
client.dispatcher().executorService().shutdown();

I considered suggesting eventSource.cancel() to shut down the executor server internally, but that might cause unexpected side effects when a client is used that is also used for non-sse activities.

So, the only suggestion I could still give, is to add this to the user documentation for the okhttp-sse module.

Good luck with your nice library, I really enjoy using it.

[daCFniel]

You saved my life 🙏. Thanks