Fixing PR comments

Author: ldecheverz-split

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

@@ -24,6 +24,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
@@ -38,7 +39,7 @@ public class PushManagerImp implements PushManager {
 
     private Future<?> _nextTokenRefreshTask;
     private final ScheduledExecutorService _scheduledExecutorService;
-    private long _expirationTime;
+    private AtomicLong _expirationTime;
 
     @VisibleForTesting
     /* package private */ PushManagerImp(AuthApiClient authApiClient,
@@ -52,6 +53,7 @@ public class PushManagerImp implements PushManager {
         _splitsWorker = splitsWorker;
         _segmentWorker = segmentWorker;
         _pushStatusTracker = pushStatusTracker;
+        _expirationTime = new AtomicLong();
         _scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder()
                 .setDaemon(true)
                 .setNameFormat("Split-SSERefreshToken-%d")
@@ -79,7 +81,7 @@ public synchronized void start() {
         AuthenticationResponse response = _authApiClient.Authenticate();
         _log.debug(String.format("Auth service response pushEnabled: %s", response.isPushEnabled()));
         if (response.isPushEnabled() && startSse(response.getToken(), response.getChannels())) {
-            _expirationTime = response.getExpiration();
+            _expirationTime.set(response.getExpiration());
             return;
         }
 
@@ -108,7 +110,7 @@ public synchronized void scheduleConnectionReset() {
             _log.debug("Starting scheduleNextTokenRefresh ...");
             stop();
             start();
-        }, _expirationTime, TimeUnit.SECONDS);
+        }, _expirationTime.get(), TimeUnit.SECONDS);
     }
 
     private boolean startSse(String token, String channels) {

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

@@ -122,6 +122,7 @@ private void startPollingMode() {
                         _pushManager.stopWorkers();
                         _pushManager.stop();
                         Thread.sleep(howLong);
+                        _incomingPushStatus.clear();
                         _pushManager.start();
                         break;
                     case STREAMING_OFF:

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

@@ -21,6 +21,7 @@
 public class EventSourceClientImp implements EventSourceClient {
     private static final Logger _log = LoggerFactory.getLogger(EventSourceClient.class);
     private static final String ERROR = "error";
+    private static final String MESSAGE = "message";
 
     private final String _baseStreamingUrl;
     private final NotificationParser _notificationParser;
@@ -101,7 +102,7 @@ private void onMessage(RawEvent event) {
             if (payload.length() > 0) {
                 _log.debug(String.format("Payload received: %s", payload));
                 switch (type) {
-                    case "message":
+                    case MESSAGE:
                         _notificationProcessor.process(_notificationParser.parseMessage(payload));
                         break;
                     case ERROR:

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

@@ -41,8 +41,7 @@ public void handleSseStatus(SSEClient.StatusMessage newStatus) {
         switch(newStatus) {
             case 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); //desactivar TODO
+                        || _sseStatus.compareAndSet(SSEClient.StatusMessage.RETRYABLE_ERROR, SSEClient.StatusMessage.CONNECTED)) {
                 }
                 break;
             case RETRYABLE_ERROR: