fixed fetchers retry until changeNumber > cahce.ChangeNumber

Author: sanzmauro

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

@@ -58,8 +58,8 @@ public void stopPeriodicFetching() {
     }
 
     @Override
-    public void refreshSplits(long targetChangeNumber) {
-        if (targetChangeNumber > _splitFetcher.changeNumber()) {
+    public synchronized void refreshSplits(long targetChangeNumber) {
+        while (targetChangeNumber > _splitFetcher.changeNumber()) {
             _splitFetcher.forceRefresh();
         }
     }
@@ -68,12 +68,13 @@ public void refreshSplits(long targetChangeNumber) {
     public void localKillSplit(String splitName, String defaultTreatment, long newChangeNumber) {
         if (newChangeNumber > _splitFetcher.changeNumber()) {
             _splitFetcher.killSplit(splitName, defaultTreatment, newChangeNumber);
+            refreshSplits(newChangeNumber);
         }
     }
 
     @Override
-    public void refreshSegment(String segmentName, long changeNumber) {
-        if (changeNumber > _segmentFetcher.getChangeNumber(segmentName)) {
+    public synchronized void refreshSegment(String segmentName, long changeNumber) {
+        while (changeNumber > _segmentFetcher.getChangeNumber(segmentName)) {
             _segmentFetcher.forceRefresh(segmentName);
         }
     }

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

@@ -103,7 +103,7 @@ private void onMessage(RawEvent event) {
         } catch (EventParsingException ex) {
             _log.debug(String.format("Error parsing the event: %s. Payload: %s", ex.getMessage(), ex.getPayload()));
         } catch (Exception e) {
-            _log.warn(String.format("Error onMessage: %s", e.getMessage()));
+            _log.debug(String.format("Error onMessage: %s", e.getMessage()));
         }
     }
 }

client/src/test/java/io/split/SplitMockServer.java

@@ -51,7 +51,9 @@ public MockResponse dispatch(RecordedRequest request) {
                     case "/api/segmentChanges/segment3?since=-1":
                         return new MockResponse().setBody(inputStreamToString("segment3.json"));
                     case "/api/segmentChanges/segment3?since=1585948850110":
-                        return new MockResponse().setBody("{\"name\": \"segment3\",\"added\": [],\"removed\": [],\"since\": 1585948850110,\"till\": 1585948850110}");
+                        return new MockResponse().setBody("{\"name\": \"segment3\",\"added\": [],\"removed\": [],\"since\": 1585948850110,\"till\": 1585948850111}");
+                    case "/api/segmentChanges/segment3?since=1585948850111":
+                        return new MockResponse().setBody("{\"name\": \"segment3\",\"added\": [],\"removed\": [],\"since\": 1585948850111,\"till\": 1585948850111}");
                     case "/api/metrics/time":
                     case "api/metrics/counter":
                         return new MockResponse().setResponseCode(200);

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

@@ -1,8 +1,12 @@
 package io.split.engine.common;
 
+import io.split.client.HttpSegmentChangeFetcher;
+import io.split.engine.SDKReadinessGates;
 import io.split.engine.experiments.RefreshableSplitFetcher;
 import io.split.engine.experiments.RefreshableSplitFetcherProvider;
 import io.split.engine.segments.RefreshableSegmentFetcher;
+import io.split.engine.segments.SegmentChangeFetcher;
+import org.apache.http.impl.client.CloseableHttpClient;
 import org.junit.Test;
 import org.mockito.Mockito;
 
@@ -56,4 +60,48 @@ public void stopPeriodicFetching() {
         Mockito.verify(refreshableSplitFetcherProvider, Mockito.times(1)).stop();
         Mockito.verify(segmentFetcher, Mockito.times(1)).stop();
     }
+
+    @Test
+    public void refreshSplits() {
+        RefreshableSplitFetcherProvider refreshableSplitFetcherProvider = Mockito.mock(RefreshableSplitFetcherProvider.class);
+        RefreshableSegmentFetcher segmentFetcher = Mockito.mock(RefreshableSegmentFetcher.class);
+        RefreshableSplitFetcher splitFetcher = Mockito.mock(RefreshableSplitFetcher.class);
+
+        Mockito.when(refreshableSplitFetcherProvider.getFetcher())
+                .thenReturn(splitFetcher);
+
+        Mockito.when(splitFetcher.changeNumber())
+                .thenReturn(450L)
+                .thenReturn(460L)
+                .thenReturn(470L)
+                .thenReturn(480L)
+                .thenReturn(500L);
+
+        Synchronizer synchronizer = new SynchronizerImp(refreshableSplitFetcherProvider, segmentFetcher);
+        synchronizer.refreshSplits(500);
+
+        Mockito.verify(splitFetcher, Mockito.times(4)).forceRefresh();
+    }
+
+    @Test
+    public void refreshSegment() {
+        RefreshableSplitFetcherProvider refreshableSplitFetcherProvider = Mockito.mock(RefreshableSplitFetcherProvider.class);
+        RefreshableSegmentFetcher segmentFetcher = Mockito.mock(RefreshableSegmentFetcher.class);
+        RefreshableSplitFetcher splitFetcher = Mockito.mock(RefreshableSplitFetcher.class);
+
+        Mockito.when(refreshableSplitFetcherProvider.getFetcher())
+                .thenReturn(splitFetcher);
+
+        Mockito.when(segmentFetcher.getChangeNumber("segment-name"))
+                .thenReturn(450L)
+                .thenReturn(460L)
+                .thenReturn(470L)
+                .thenReturn(480L)
+                .thenReturn(500L);
+
+        Synchronizer synchronizer = new SynchronizerImp(refreshableSplitFetcherProvider, segmentFetcher);
+        synchronizer.refreshSegment("segment-name", 500);
+
+        Mockito.verify(segmentFetcher, Mockito.times(4)).forceRefresh("segment-name");
+    }
 }

client/src/test/java/io/split/engine/sse/workers/SplitsWorkerTest.java

@@ -60,7 +60,7 @@ public void killShouldTriggerFetch() {
     }
 
     @Test
-    public void messagesNotProcesedWhenWorkerStopped() throws InterruptedException {
+    public void messagesNotProcessedWhenWorkerStopped() throws InterruptedException {
         Synchronizer syncMock = Mockito.mock(Synchronizer.class);
         SplitsWorker splitsWorker = new SplitsWorkerImp(syncMock);
         splitsWorker.start();