handle more cases and add some more integration tests

Author: mredolatti

client/src/main/java/io/split/engine/sse/client/SSEClient.java

@@ -16,7 +16,6 @@
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.net.SocketException;
-import java.net.SocketTimeoutException;
 import java.net.URI;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -40,7 +39,7 @@ private enum ConnectionState {
         CLOSED
     }
 
-
+    private final static String SOCKET_CLOSED_MESSAGE = "Socket closed";
     private final static String KEEP_ALIVE_PAYLOAD = ":keepalive\n";
     private final static Integer CONNECT_TIMEOUT = 30000;
     private final static Integer SOCKET_TIMEOUT = 70000;
@@ -110,29 +109,21 @@ private void connectAndLoop(URI uri, CountDownLatch signal) {
             while (isOpen() && !Thread.currentThread().isInterrupted()) {
                 try {
                     handleMessage(readMessageAsString(reader));
-                } catch (EOFException | SocketTimeoutException exc) {
-                    // EOFException: This is when ably closes the connection on their end. IE: an invalid or expired token.
-                    // SocketTimeoutException: KeepAlive expired
-                    _log.warn(exc.getMessage());
-                    _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
-                    return;
                 } catch (SocketException exc) {
                     _log.warn(exc.getMessage());
-
-                    if ("Socket closed".equals(exc.getMessage())) {
-                        // Connection closed by us
+                    if (SOCKET_CLOSED_MESSAGE.equals(exc.getMessage())) { // Connection closed by us
                         _statusCallback.apply(StatusMessage.FORCED_STOP);
                         return;
                     }
-
                     // Connection closed by server
                     _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
-                    return;
+                } catch (IOException exc) { // Other type of connection error
+                    _log.warn(exc.getMessage());
+                    _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
                 }
             }
-        } catch (Exception e) {
-            _log.error(e.getMessage());
-
+        } catch (Exception e) { // Any other error non related to the connection disables streaming altogether
+            _log.error(e.getMessage(), e);
             _statusCallback.apply(StatusMessage.NONRETRYABLE_ERROR);
         } finally {
             try {

client/src/test/java/io/split/SSEMockServer.java

@@ -4,6 +4,7 @@
 import org.glassfish.grizzly.utils.Pair;
 import org.glassfish.hk2.utilities.binding.AbstractBinder;
 import org.glassfish.jersey.grizzly2.httpserver.GrizzlyHttpServerFactory;
+import org.glassfish.jersey.media.sse.OutboundEvent;
 import org.glassfish.jersey.server.ResourceConfig;
 import org.jvnet.hk2.annotations.Service;
 
@@ -28,9 +29,12 @@ public class SSEMockServer {
     private static final String BASE_URL = "http://localhost:%d";
     private final SseEventQueue _queue;
     private final Validator _validator;
-    private AtomicInteger _port;
+    private final AtomicInteger _port;
     private HttpServer _server;
 
+    public static final OutboundEvent CONNECTION_CLOSED_BY_REMOTE_HOST = new OutboundEvent.Builder().comment("CCBRH").build();
+//    public static final OutboundEvent CONNECTION_RESET_EVENT = new OutboundEvent.Builder().comment("RST").build();
+
     public SSEMockServer(SseEventQueue queue, Validator validator) {
         _queue = queue;
         _validator = validator;
@@ -110,6 +114,10 @@ public void getServerSentEvents(@Context SseEventSink eventSink,
                 while(!eventSink.isClosed()) {
                     try {
                         OutboundSseEvent event = _eventsToSend.pull();
+                        if (CONNECTION_CLOSED_BY_REMOTE_HOST == event) { // Comparing references, no need for .equals()
+                            eventSink.close();
+                            return;
+                        }
                         eventSink.send(event);
                     } catch (InterruptedException e) {
                         break;

client/src/test/java/io/split/client/SplitClientIntegrationTest.java

@@ -500,6 +500,50 @@ public void keepAlive() throws IOException, TimeoutException, InterruptedExcepti
         sseServer.stop();
     }
 
+    @Test
+    public void testConnectionClosedByRemoteHostIsProperlyHandled() throws IOException, TimeoutException, InterruptedException, URISyntaxException {
+        SplitMockServer splitServer = new SplitMockServer();
+        SSEMockServer.SseEventQueue eventQueue = new SSEMockServer.SseEventQueue();
+        SSEMockServer sseServer = buildSSEMockServer(eventQueue);
+
+        splitServer.start();
+        sseServer.start();
+
+        SplitClientConfig config = buildSplitClientConfig("enabled", splitServer.getUrl(), sseServer.getPort(), true, 50);
+        SplitFactory factory = SplitFactoryBuilder.build("fake-api-token-1", config);
+        SplitClient client = factory.client();
+        client.blockUntilReady();
+
+        String result = client.getTreatment("admin", "push_test");
+        Assert.assertEquals("on_whitelist", result);
+        eventQueue.push(SSEMockServer.CONNECTION_CLOSED_BY_REMOTE_HOST);
+        Thread.sleep(1000);
+        result = client.getTreatment("admin", "push_test");
+        Assert.assertEquals("on_whitelist", result);
+    }
+
+    @Test
+    public void testConnectionClosedIsProperlyHandled() throws IOException, TimeoutException, InterruptedException, URISyntaxException {
+        SplitMockServer splitServer = new SplitMockServer();
+        SSEMockServer.SseEventQueue eventQueue = new SSEMockServer.SseEventQueue();
+        SSEMockServer sseServer = buildSSEMockServer(eventQueue);
+
+        splitServer.start();
+        sseServer.start();
+
+        SplitClientConfig config = buildSplitClientConfig("enabled", splitServer.getUrl(), sseServer.getPort(), true, 50);
+        SplitFactory factory = SplitFactoryBuilder.build("fake-api-token-1", config);
+        SplitClient client = factory.client();
+        client.blockUntilReady();
+
+        String result = client.getTreatment("admin", "push_test");
+        Assert.assertEquals("on_whitelist", result);
+        sseServer.stop();
+        Thread.sleep(1000);
+        result = client.getTreatment("admin", "push_test");
+        Assert.assertEquals("on_whitelist", result);
+    }
+    
     private SSEMockServer buildSSEMockServer(SSEMockServer.SseEventQueue eventQueue) {
         return new SSEMockServer(eventQueue, (token, version, channel) -> {
             if (!"1.1".equals(version)) {