added new sseClient status: FORCED_STOP

Author: sanzmauro

client/src/main/java/io/split/engine/common/SyncManagerImp.java

@@ -110,7 +110,7 @@ private void startPollingMode() {
                         _pushManager.start();
                         break;
                     case STREAMING_OFF:
-                        _pushManager.stop();
+                        _pushManager.stopWorkers();
                         _synchronizer.startPeriodicFetching();
                         if (null != _pushStatusMonitorTask) {
                             _pushStatusMonitorTask.cancel(false);

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

@@ -72,6 +72,6 @@ private AuthenticationResponse getSuccessResponse(String jsonContent) {
             expiration = response.getExpiration();
         }
 
-        return new AuthenticationResponse(response.isPushEnabled(), response.getToken(), channels, 3000/*expiration*/, false);
+        return new AuthenticationResponse(response.isPushEnabled(), response.getToken(), channels, expiration, false);
     }
 }

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

@@ -33,8 +33,8 @@ public synchronized void reset() {
 
     @Override
     public void handleSseStatus(SSEClient.StatusMessage newStatus) {
-        _log.debug(String.format("handleSseStatus new status: %s", newStatus.toString()));
-        _log.debug(String.format("handleSseStatus current status: %s", _sseStatus.get().toString()));
+        _log.debug(String.format("Current status: %s. New status: %s", _sseStatus.get().toString(), newStatus.toString()));
+
         switch(newStatus) {
             case CONNECTED:
                 if (_sseStatus.compareAndSet(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS, SSEClient.StatusMessage.CONNECTED)
@@ -53,6 +53,13 @@ public void handleSseStatus(SSEClient.StatusMessage newStatus) {
                     _statusMessages.offer(PushManager.Status.STREAMING_OFF);
                 }
                 break;
+            case FORCED_STOP:
+                if (_sseStatus.compareAndSet(SSEClient.StatusMessage.CONNECTED, SSEClient.StatusMessage.FORCED_STOP)
+                    || _sseStatus.compareAndSet(SSEClient.StatusMessage.RETRYABLE_ERROR, SSEClient.StatusMessage.FORCED_STOP)
+                    || _sseStatus.compareAndSet(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS, SSEClient.StatusMessage.FORCED_STOP)) {
+                    _statusMessages.offer(PushManager.Status.STREAMING_DOWN);
+                }
+                break;
             case INITIALIZATION_IN_PROGRESS: // Restore initial status
                 reset();
                 break;
@@ -62,6 +69,7 @@ public void handleSseStatus(SSEClient.StatusMessage newStatus) {
     @Override
     public void handleIncomingControlEvent(ControlNotification controlNotification) {
         _log.debug(String.format("handleIncomingOccupancyEvent: %s", controlNotification.getControlType()));
+
         if (_backendStatus.get().equals(ControlType.STREAMING_DISABLED)) {
             return;
         }
@@ -88,6 +96,7 @@ public void handleIncomingControlEvent(ControlNotification controlNotification)
     @Override
     public void handleIncomingOccupancyEvent(OccupancyNotification occupancyNotification) {
         _log.debug(String.format("handleIncomingOccupancyEvent: publishers=%d", occupancyNotification.getMetrics().getPublishers()));
+
         int publishers = occupancyNotification.getMetrics().getPublishers();
         if (publishers <= 0 && _publishersOnline.compareAndSet(true, false) && _backendStatus.get().equals(ControlType.STREAMING_RESUMED)) {
             _statusMessages.offer(PushManager.Status.STREAMING_DOWN);
@@ -99,6 +108,7 @@ public void handleIncomingOccupancyEvent(OccupancyNotification occupancyNotifica
     @Override
     public void handleIncomingAblyError(ErrorNotification notification) {
         _log.debug(String.format("handleIncomingAblyError: %s", notification.getMessage()));
+
         if (_backendStatus.get().equals(ControlType.STREAMING_DISABLED)) {
             return; // Ignore
         }
@@ -113,6 +123,7 @@ public void handleIncomingAblyError(ErrorNotification notification) {
     @Override
     public synchronized void forcePushDisable() {
         _log.debug("forcePushDisable");
+
         _publishersOnline.set(false);
         _sseStatus.set(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS);
         _backendStatus.set(ControlType.STREAMING_DISABLED);

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

@@ -1,7 +1,6 @@
 package io.split.engine.sse.client;
 
 import com.google.common.base.Strings;
-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 +31,8 @@ public enum StatusMessage {
         CONNECTED,
         RETRYABLE_ERROR,
         NONRETRYABLE_ERROR,
-        INITIALIZATION_IN_PROGRESS
+        INITIALIZATION_IN_PROGRESS,
+        FORCED_STOP
     }
 
     private enum ConnectionState {
@@ -117,7 +117,7 @@ private void connectAndLoop(URI uri, CountDownLatch signal) {
                     _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
                 } catch (SocketException exc) { // Connection closed by us
                     if ("Socket closed".equals(exc.getMessage())) {
-                        _statusCallback.apply(StatusMessage.NONRETRYABLE_ERROR);
+                        _statusCallback.apply(StatusMessage.FORCED_STOP);
                         return;
                     }