Pulling develompent's changes

Author: ldecheverz-split

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

@@ -62,7 +62,6 @@ public static PushManagerImp build(Synchronizer synchronizer,
                                        String streamingUrl,
                                        String authUrl,
                                        CloseableHttpClient httpClient,
-                                       int authRetryBackOffBase,
                                        LinkedBlockingQueue<PushManager.Status> statusMessages,
                                        CloseableHttpClient sseHttpClient) {
         SplitsWorker splitsWorker = new SplitsWorkerImp(synchronizer);
@@ -104,7 +103,6 @@ public synchronized void stop() {
 
     @Override
     public synchronized void scheduleConnectionReset() {
-        _expirationTime = 120l;
         _log.debug(String.format("scheduleNextTokenRefresh in %s SECONDS", _expirationTime));
         _nextTokenRefreshTask = _scheduledExecutorService.schedule(() -> {
             _log.debug("Starting scheduleNextTokenRefresh ...");

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

@@ -62,7 +62,7 @@ public static SyncManagerImp build(boolean streamingEnabledConfig,
                                        SegmentCache segmentCache) {
         LinkedBlockingQueue<PushManager.Status> pushMessages = new LinkedBlockingQueue<>();
         Synchronizer synchronizer = new SynchronizerImp(splitSynchronizationTask, splitFetcher, segmentSynchronizationTaskImp, splitCache, segmentCache);
-        PushManager pushManager = PushManagerImp.build(synchronizer, streamingServiceUrl, authUrl, httpClient, authRetryBackOffBase, pushMessages, sseHttpClient);
+        PushManager pushManager = PushManagerImp.build(synchronizer, streamingServiceUrl, authUrl, httpClient, pushMessages, sseHttpClient);
         return new SyncManagerImp(streamingEnabledConfig, synchronizer, pushManager, pushMessages, authRetryBackOffBase);
     }
 

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

@@ -144,6 +144,7 @@ public void notifyStreamingReady() {
     @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

@@ -131,9 +131,11 @@ private void connectAndLoop(URI uri, CountDownLatch signal) {
                     _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
                     return;
                 } catch (IOException exc) { // Other type of connection error
-                    _log.info(String.format("SSE connection ended abruptly: %s. Retrying", exc.getMessage()));
-                    _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
-                    return;
+                    if(!_forcedStop.get()) {
+                        _log.debug(String.format("SSE connection ended abruptly: %s. Retying", exc.getMessage()));
+                        _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
+                        return;
+                    }
                 }
             }
         } catch (Exception e) { // Any other error non related to the connection disables streaming altogether