Adding TelemtrySyncTask

Author: ldecheverz-split

client/src/main/java/io/split/telemetry/synchronizer/HttpTelemetryMemorySender.java

@@ -1,14 +1,11 @@
 package io.split.telemetry.synchronizer;
 
 import com.google.common.annotations.VisibleForTesting;
-import io.split.client.impressions.HttpImpressionsSender;
 import io.split.client.utils.Utils;
+import io.split.service.HttpPostImp;
 import io.split.telemetry.domain.Config;
 import io.split.telemetry.domain.Stats;
-import io.split.service.HttpPostImp;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -20,8 +17,6 @@ public class HttpTelemetryMemorySender extends HttpPostImp {
     private static final String CONFIG_METRICS = "Config metrics ";
     private static final String STATS_METRICS = "Stats metrics ";
 
-    private static final Logger _logger = LoggerFactory.getLogger(HttpImpressionsSender.class);
-
     private final URI _impressionConfigTarget;
     private final URI _impressionStatsTarget;
 

…nchronizer/TelemetrySynchronizerImp.java …try/synchronizer/SynchronizerMemory.javaclient/src/main/java/io/split/telemetry/synchronizer/TelemetrySynchronizerImp.java renamed to client/src/main/java/io/split/telemetry/synchronizer/SynchronizerMemory.java client/src/main/java/io/split/telemetry/synchronizer/TelemetrySynchronizerImp.java renamed to client/src/main/java/io/split/telemetry/synchronizer/SynchronizerMemory.java

@@ -19,12 +19,10 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
-public class TelemetrySynchronizerImp implements TelemetrySynchronizer{
+public class SynchronizerMemory implements TelemetrySynchronizer{
 
     private static final int OPERATION_MODE = 0;
     private static  final String STORAGE = "memory";
@@ -34,8 +32,8 @@ public class TelemetrySynchronizerImp implements TelemetrySynchronizer{
     private SplitCache _splitCache;
     private SegmentCache _segmentCache;
 
-    public TelemetrySynchronizerImp(CloseableHttpClient client, URI telemetryRootEndpoint, TelemetryStorageConsumer telemetryStorageConsumer, SplitCache splitCache,
-            SegmentCache segmentCache) throws URISyntaxException {
+    public SynchronizerMemory(CloseableHttpClient client, URI telemetryRootEndpoint, TelemetryStorageConsumer telemetryStorageConsumer, SplitCache splitCache,
+                              SegmentCache segmentCache) throws URISyntaxException {
         _httpHttpTelemetryMemorySender = HttpTelemetryMemorySender.create(client, telemetryRootEndpoint);
         _teleTelemetryStorageConsumer = telemetryStorageConsumer;
         _splitCache = splitCache;

client/src/main/java/io/split/telemetry/synchronizer/TelemetrySyncTask.java

@@ -0,0 +1,37 @@
+package io.split.telemetry.synchronizer;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+public class TelemetrySyncTask {
+
+    private final ScheduledExecutorService _telemetrySyncScheduledExecutorService;
+    private final TelemetrySynchronizer _telemetrySynchronizer;
+    private final int _telemetryRefreshRate;
+
+    public TelemetrySyncTask(int telemetryRefreshRate, TelemetrySynchronizer telemetrySynchronizer) {
+        ThreadFactory telemetrySyncThreadFactory = new ThreadFactoryBuilder()
+                .setDaemon(true)
+                .setNameFormat("Telemetry-synk-%d")
+                .build();
+        _telemetrySynchronizer = telemetrySynchronizer; //TODO
+        _telemetryRefreshRate = telemetryRefreshRate;
+        _telemetrySyncScheduledExecutorService = Executors.newSingleThreadScheduledExecutor(telemetrySyncThreadFactory);
+    }
+
+    @VisibleForTesting
+    protected void startScheduledTask() throws Exception {
+        _telemetrySyncScheduledExecutorService.scheduleWithFixedDelay(() -> {
+            try {
+                _telemetrySynchronizer.synchronizeStats();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        },0l,  _telemetryRefreshRate, TimeUnit.SECONDS);
+    }
+}

…onizer/TelemetrySynchronizerImpTest.java …synchronizer/SynchronizerMemoryTest.javaclient/src/test/java/io/split/telemetry/synchronizer/TelemetrySynchronizerImpTest.java renamed to client/src/test/java/io/split/telemetry/synchronizer/SynchronizerMemoryTest.java client/src/test/java/io/split/telemetry/synchronizer/TelemetrySynchronizerImpTest.java renamed to client/src/test/java/io/split/telemetry/synchronizer/SynchronizerMemoryTest.java

@@ -1,21 +1,15 @@
 package io.split.telemetry.synchronizer;
 
 import io.split.TestHelper;
-import io.split.cache.InMemoryCacheImp;
 import io.split.cache.SegmentCache;
 import io.split.cache.SegmentCacheInMemoryImpl;
 import io.split.cache.SplitCache;
 import io.split.client.SplitClientConfig;
-import io.split.service.HttpPostImp;
 import io.split.telemetry.storage.InMemoryTelemetryStorage;
 import io.split.telemetry.storage.TelemetryStorageConsumer;
-import org.apache.hc.client5.http.classic.methods.HttpOptions;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
-import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
-import org.apache.hc.core5.http.ClassicHttpRequest;
 import org.apache.hc.core5.http.HttpStatus;
 import org.junit.Test;
-import org.mockito.Mock;
 import org.mockito.Mockito;
 
 import java.io.IOException;
@@ -25,7 +19,7 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 
-public class TelemetrySynchronizerImpTest{
+public class SynchronizerMemoryTest {
 
     public static final String TELEMETRY_ENDPOINT = "https://telemetry.split.io/api/v1";
 
@@ -53,7 +47,7 @@ private TelemetrySynchronizer getTelemetrySynchronizer(CloseableHttpClient httpC
         TelemetryStorageConsumer consumer = Mockito.mock(InMemoryTelemetryStorage.class);
         SplitCache splitCache = Mockito.mock(SplitCache.class);
         SegmentCache segmentCache = Mockito.mock(SegmentCacheInMemoryImpl.class);
-        TelemetrySynchronizer telemetrySynchronizer = new TelemetrySynchronizerImp(httpClient, URI.create(TELEMETRY_ENDPOINT), consumer, splitCache, segmentCache);
+        TelemetrySynchronizer telemetrySynchronizer = new SynchronizerMemory(httpClient, URI.create(TELEMETRY_ENDPOINT), consumer, splitCache, segmentCache);
         return telemetrySynchronizer;
     }
 

client/src/test/java/io/split/telemetry/synchronizer/TelemetrySyncTaskTest.java

@@ -0,0 +1,17 @@
+package io.split.telemetry.synchronizer;
+
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class TelemetrySyncTaskTest {
+
+    @Test
+    public void testSynchronizationTask() throws Exception {
+        TelemetrySynchronizer telemetrySynchronizer = Mockito.mock(SynchronizerMemory.class);
+        TelemetrySyncTask telemetrySyncTask = new TelemetrySyncTask(1, telemetrySynchronizer);
+        telemetrySyncTask.startScheduledTask();
+        Thread.sleep(3000);
+        Mockito.verify(telemetrySynchronizer, Mockito.times(3)).synchronizeStats();
+    }
+
+}