Update PushManager and SyncManager

Author: nmayorsplit

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

@@ -102,7 +102,7 @@ public synchronized void start() {
             return;
         }
 
-        stop();
+        cleanUpResources();
         if (response.isRetry()) {
             _pushStatusTracker.handleSseStatus(SSEClient.StatusMessage.RETRYABLE_ERROR);
         } else {
@@ -113,16 +113,11 @@ public synchronized void start() {
     @Override
     public synchronized void stop() {
         _log.debug("Stopping PushManagerImp");
-        _eventSourceClient.stop();
-        stopWorkers();
-        if (_nextTokenRefreshTask != null) {
-            _log.debug("Cancel nextTokenRefreshTask");
-            _nextTokenRefreshTask.cancel(false);
-        }
+        cleanUpResources();
     }
 
     @Override
-    public synchronized void scheduleConnectionReset() {
+    public void scheduleConnectionReset() {
         _log.debug(String.format("scheduleNextTokenRefresh in %s SECONDS", _expirationTime));
         _nextTokenRefreshTask = _scheduledExecutorService.schedule(() -> {
             _log.debug("Starting scheduleNextTokenRefresh ...");
@@ -142,14 +137,23 @@ private boolean startSse(String token, String channels) {
     }
 
     @Override
-    public synchronized void startWorkers() {
+    public void startWorkers() {
         _featureFlagsWorker.start();
         _segmentWorker.start();
     }
 
     @Override
-    public synchronized void stopWorkers() {
+    public void stopWorkers() {
         _featureFlagsWorker.stop();
         _segmentWorker.stop();
     }
+
+    private void cleanUpResources() {
+        _eventSourceClient.stop();
+        stopWorkers();
+        if (_nextTokenRefreshTask != null) {
+            _log.debug("Cancel nextTokenRefreshTask");
+            _nextTokenRefreshTask.cancel(false);
+        }
+    }
 }

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

@@ -218,7 +218,6 @@ private void startPollingMode() {
                         long howLong = _backoff.interval();
                         _log.info(String.format("Retryable error in streaming subsystem. Switching to polling and retrying in %d seconds", howLong));
                         _synchronizer.startPeriodicFetching();
-                        _pushManager.stopWorkers();
                         _pushManager.stop();
                         Thread.sleep(howLong * 1000);
                         _incomingPushStatus.clear();

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

@@ -22,18 +22,22 @@ public class PushManagerTest {
     private PushManager _pushManager;
     private PushStatusTracker _pushStatusTracker;
     private TelemetryStorage _telemetryStorage;
+    private FeatureFlagsWorker _featureFlagsWorker;
+    private SegmentsWorkerImp _segmentsWorkerImp;
 
     @Before
     public void setUp() {
+        _featureFlagsWorker = Mockito.mock(FeatureFlagsWorker.class);
+        _segmentsWorkerImp = Mockito.mock(SegmentsWorkerImp.class);
         _authApiClient = Mockito.mock(AuthApiClient.class);
         _eventSourceClient = Mockito.mock(EventSourceClient.class);
         _backoff = Mockito.mock(Backoff.class);
         _pushStatusTracker = Mockito.mock(PushStatusTrackerImp.class);
         _telemetryStorage = new InMemoryTelemetryStorage();
         _pushManager = new PushManagerImp(_authApiClient,
                 _eventSourceClient,
-                Mockito.mock(FeatureFlagsWorker.class),
-                Mockito.mock(SegmentsWorkerImp.class),
+                _featureFlagsWorker,
+                _segmentsWorkerImp,
                 _pushStatusTracker,
                 _telemetryStorage,
                 null);
@@ -107,4 +111,60 @@ public void startWithPushDisabledAndRetryTrueShouldConnect() throws InterruptedE
         Thread.sleep(1500);
         Mockito.verify(_pushStatusTracker, Mockito.times(1)).handleSseStatus(SSEClient.StatusMessage.RETRYABLE_ERROR);
     }
+
+
+    @Test
+    public void startAndStop() throws InterruptedException {
+        AuthenticationResponse response = new AuthenticationResponse(true, "token-test", "channels-test", 1, false);
+
+        Mockito.when(_authApiClient.Authenticate())
+                .thenReturn(response);
+
+        Mockito.when(_eventSourceClient.start(response.getChannels(), response.getToken()))
+                .thenReturn(true);
+
+        _pushManager.start();
+
+        Mockito.verify(_authApiClient, Mockito.times(1)).Authenticate();
+        Mockito.verify(_eventSourceClient, Mockito.times(1)).start(response.getChannels(), response.getToken());
+
+        Thread.sleep(1500);
+
+        Mockito.verify(_pushStatusTracker, Mockito.times(0)).handleSseStatus(SSEClient.StatusMessage.RETRYABLE_ERROR);
+        Mockito.verify(_pushStatusTracker, Mockito.times(0)).forcePushDisable();
+        Assert.assertEquals(1, _telemetryStorage.popStreamingEvents().size());
+
+        _pushManager.stop();
+
+        Mockito.verify(_eventSourceClient, Mockito.times(1)).stop();
+        Mockito.verify(_featureFlagsWorker, Mockito.times(1)).stop();
+        Mockito.verify(_segmentsWorkerImp, Mockito.times(1)).stop();
+    }
+
+    @Test
+    public void validateStartWorkers() {
+        _pushManager.startWorkers();
+        Mockito.verify(_featureFlagsWorker, Mockito.times(1)).start();
+        Mockito.verify(_segmentsWorkerImp, Mockito.times(1)).start();
+    }
+
+    @Test
+    public void validateScheduleConnectionReset() throws InterruptedException {
+        AuthenticationResponse response = new AuthenticationResponse(false, "token-test", "channels-test", 3, false);
+
+        Mockito.when(_authApiClient.Authenticate())
+                .thenReturn(response);
+
+        Mockito.when(_eventSourceClient.start(response.getChannels(), response.getToken()))
+                .thenReturn(true);
+
+        _pushManager.start();
+
+        _pushManager.scheduleConnectionReset();
+        Thread.sleep(1000);
+
+        Mockito.verify(_eventSourceClient, Mockito.times(3)).stop();
+        Mockito.verify(_featureFlagsWorker, Mockito.times(3)).stop();
+        Mockito.verify(_segmentsWorkerImp, Mockito.times(3)).stop();
+    }
 }