FIxing PR comments

Author: ldecheverz-split

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

@@ -8,6 +8,7 @@
 import io.split.engine.sse.EventSourceClientImp;
 import io.split.engine.sse.PushStatusTracker;
 import io.split.engine.sse.PushStatusTrackerImp;
+import io.split.engine.sse.client.SSEClient;
 import io.split.engine.sse.dtos.AuthenticationResponse;
 import io.split.engine.sse.dtos.SegmentQueueDto;
 import io.split.engine.sse.workers.SegmentsWorkerImp;
@@ -87,7 +88,7 @@ public synchronized void start() {
 
         stop();
         if (response.isRetry()) {
-            _pushStatusTracker.forceRetryableError();//retriable error
+            _pushStatusTracker.handleSseStatus(SSEClient.StatusMessage.RETRYABLE_ERROR);
         } else {
             _pushStatusTracker.forcePushDisable();
         }

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

@@ -97,7 +97,7 @@ private void onMessage(RawEvent event) {
             String type = event.event();
             String payload = event.data();
             if(_firstEvent.compareAndSet(false, true) && !ERROR.equals(type)){
-                _pushStatusTracker.notifyStreamingReady();
+                _pushStatusTracker.handleSseStatus(SSEClient.StatusMessage.FIRST_EVENT);
             }
             if (payload.length() > 0) {
                 _log.debug(String.format("Payload received: %s", payload));

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

@@ -11,6 +11,4 @@ public interface PushStatusTracker {
     void handleIncomingAblyError(ErrorNotification notification);
     void handleSseStatus(SSEClient.StatusMessage newStatus);
     void forcePushDisable();
-    void notifyStreamingReady();
-    void forceRetryableError();
 }

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

@@ -28,7 +28,7 @@ public PushStatusTrackerImp(LinkedBlockingQueue<PushManager.Status> statusMessag
         _statusMessages = statusMessages;
     }
 
-    public synchronized void reset() {
+    private synchronized void reset() {
         _publishersOnline.set(true);
         _sseStatus.set(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS);
         _backendStatus.set(ControlType.STREAMING_RESUMED);
@@ -39,10 +39,12 @@ public void handleSseStatus(SSEClient.StatusMessage newStatus) {
         _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)
-                        || _sseStatus.compareAndSet(SSEClient.StatusMessage.RETRYABLE_ERROR, SSEClient.StatusMessage.CONNECTED)) {
+            case FIRST_EVENT:
+                if (SSEClient.StatusMessage.CONNECTED.equals(_sseStatus.get())) {
+                    _statusMessages.offer(PushManager.Status.STREAMING_READY);
                 }
+            case CONNECTED:
+                _sseStatus.compareAndSet(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS, SSEClient.StatusMessage.CONNECTED);
                 break;
             case RETRYABLE_ERROR:
                 if (_sseStatus.compareAndSet(SSEClient.StatusMessage.CONNECTED, SSEClient.StatusMessage.RETRYABLE_ERROR)) {
@@ -134,16 +136,6 @@ public synchronized void forcePushDisable() {
         _backendStatus.set(ControlType.STREAMING_DISABLED);
         _statusMessages.offer(PushManager.Status.STREAMING_OFF);
     }
-
-    @Override
-    public void notifyStreamingReady() {
-        _statusMessages.offer(PushManager.Status.STREAMING_READY);
-    }
-
-    @Override
-    public void forceRetryableError() {
-        _statusMessages.offer(PushManager.Status.STREAMING_BACKOFF);
-    }
 
     private boolean isPublishers() {
         for(Integer publisher : regions.values()) {

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

@@ -32,7 +32,8 @@ public enum StatusMessage {
         RETRYABLE_ERROR,
         NONRETRYABLE_ERROR,
         INITIALIZATION_IN_PROGRESS,
-        FORCED_STOP
+        FORCED_STOP,
+        FIRST_EVENT
     }
 
     private enum ConnectionState {

client/src/test/java/io/split/engine/common/PushManagerTest.java

@@ -4,6 +4,7 @@
 import io.split.engine.sse.EventSourceClient;
 import io.split.engine.sse.PushStatusTracker;
 import io.split.engine.sse.PushStatusTrackerImp;
+import io.split.engine.sse.client.SSEClient;
 import io.split.engine.sse.dtos.AuthenticationResponse;
 import io.split.engine.sse.workers.SegmentsWorkerImp;
 import io.split.engine.sse.workers.SplitsWorker;
@@ -55,7 +56,7 @@ public void startWithPushEnabledShouldConnect() throws InterruptedException {
 
         Thread.sleep(1500);
 
-        Mockito.verify(_pushStatusTracker, Mockito.times(0)).forceRetryableError();
+        Mockito.verify(_pushStatusTracker, Mockito.times(0)).handleSseStatus(SSEClient.StatusMessage.RETRYABLE_ERROR);
         Mockito.verify(_pushStatusTracker, Mockito.times(0)).forcePushDisable();
     }
 
@@ -98,6 +99,6 @@ public void startWithPushDisabledAndRetryTrueShouldConnect() throws InterruptedE
         Mockito.verify(_eventSourceClient, Mockito.times(1)).stop();
 
         Thread.sleep(1500);
-        Mockito.verify(_pushStatusTracker, Mockito.times(1)).forceRetryableError();
+        Mockito.verify(_pushStatusTracker, Mockito.times(1)).handleSseStatus(SSEClient.StatusMessage.RETRYABLE_ERROR);
     }
 }