sseClient improvements

Author: sanzmauro

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

@@ -17,7 +17,7 @@ public class PushStatusTrackerImp implements PushStatusTracker {
     private static final Logger _log = LoggerFactory.getLogger(PushStatusTracker.class);
 
     private final AtomicBoolean _publishersOnline = new AtomicBoolean(true);
-    private final AtomicReference<SSEClient.StatusMessage> _sseStatus = new AtomicReference<>(SSEClient.StatusMessage.DISCONNECTED);
+    private final AtomicReference<SSEClient.StatusMessage> _sseStatus = new AtomicReference<>(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS);
     private final AtomicReference<ControlType> _backendStatus = new AtomicReference<>(ControlType.STREAMING_RESUMED);
     private final LinkedBlockingQueue<PushManager.Status> _statusMessages;
 
@@ -27,7 +27,7 @@ public PushStatusTrackerImp(LinkedBlockingQueue<PushManager.Status> statusMessag
 
     public synchronized void reset() {
         _publishersOnline.set(true);
-        _sseStatus.set(SSEClient.StatusMessage.DISCONNECTED);
+        _sseStatus.set(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS);
         _backendStatus.set(ControlType.STREAMING_RESUMED);
     }
 
@@ -37,7 +37,7 @@ public void handleSseStatus(SSEClient.StatusMessage newStatus) {
         _log.debug(String.format("handleSseStatus current status: %s", _sseStatus.get().toString()));
         switch(newStatus) {
             case CONNECTED:
-                if (_sseStatus.compareAndSet(SSEClient.StatusMessage.DISCONNECTED, SSEClient.StatusMessage.CONNECTED)
+                if (_sseStatus.compareAndSet(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS, SSEClient.StatusMessage.CONNECTED)
                     || _sseStatus.compareAndSet(SSEClient.StatusMessage.RETRYABLE_ERROR, SSEClient.StatusMessage.CONNECTED)) {
                     _statusMessages.offer(PushManager.Status.STREAMING_READY);
                 }
@@ -53,7 +53,7 @@ public void handleSseStatus(SSEClient.StatusMessage newStatus) {
                     _statusMessages.offer(PushManager.Status.STREAMING_OFF);
                 }
                 break;
-            case DISCONNECTED: // Restore initial status
+            case INITIALIZATION_IN_PROGRESS: // Restore initial status
                 reset();
                 break;
         }
@@ -114,7 +114,7 @@ public void handleIncomingAblyError(ErrorNotification notification) {
     public synchronized void forcePushDisable() {
         _log.debug("forcePushDisable");
         _publishersOnline.set(false);
-        _sseStatus.set(SSEClient.StatusMessage.DISCONNECTED);
+        _sseStatus.set(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS);
         _backendStatus.set(ControlType.STREAMING_DISABLED);
         _statusMessages.offer(PushManager.Status.STREAMING_OFF);
     }

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

@@ -1,7 +1,7 @@
 package io.split.engine.sse.client;
 
 import com.google.common.base.Strings;
-import io.split.engine.sse.EventSourceClient;
+import io.split.engine.sse.exceptions.EventParsingException;
 import org.apache.http.client.config.RequestConfig;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpGet;
@@ -32,7 +32,7 @@ public enum StatusMessage {
         CONNECTED,
         RETRYABLE_ERROR,
         NONRETRYABLE_ERROR,
-        DISCONNECTED
+        INITIALIZATION_IN_PROGRESS
     }
 
     private enum ConnectionState {
@@ -63,6 +63,8 @@ public synchronized boolean open(URI uri) {
             return false;
         }
 
+        _statusCallback.apply(StatusMessage.INITIALIZATION_IN_PROGRESS);
+
         CountDownLatch signal = new CountDownLatch(1);
         Thread thread = new Thread(() -> connectAndLoop(uri, signal));
         thread.setDaemon(true);
@@ -109,29 +111,31 @@ private void connectAndLoop(URI uri, CountDownLatch signal) {
                     handleMessage(readMessageAsString(reader));
                 } catch (EOFException exc) {
                     // This is when ably closes the connection on their end. IE: an invalid or expired token.
-                    // Evaluate if we should send the DISCONNECTED event or not.
-                    _statusCallback.apply(StatusMessage.DISCONNECTED);
+                    _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
                     return;
                 } catch (SocketTimeoutException exc) { // KeepAlive expired
                     _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
                 } catch (SocketException exc) { // Connection closed by us
                     if ("Socket closed".equals(exc.getMessage())) {
-                        _statusCallback.apply(StatusMessage.DISCONNECTED);
+                        _statusCallback.apply(StatusMessage.NONRETRYABLE_ERROR);
                         return;
                     }
+
                     throw exc; // If it's not a socket closed (caused by us), rethrow the exception
                 }
             }
-        } catch (IOException e) {
-            _log.warn(e.getMessage());
+        } catch (Exception e) {
+            _log.error(e.getMessage());
             _statusCallback.apply(StatusMessage.NONRETRYABLE_ERROR);
-        } finally {
+        }
+        finally {
             try {
                 _ongoingResponse.get().close();
             } catch (IOException e) {
                 _log.warn(e.getMessage());
             }
 
+            _state.set(ConnectionState.CLOSED);
             _log.warn("SSEClient finished.");
         }
     }