Extract impressions module

client/pom.xml

@@ -172,6 +172,11 @@
             <artifactId>tracker</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>io.split.client</groupId>
+            <artifactId>impressions</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.split.client</groupId>
             <artifactId>targeting-engine</artifactId>

client/src/main/java/io/split/client/SplitFactoryImpl.java

@@ -17,8 +17,12 @@
 import io.split.client.impressions.ImpressionListener;
 import io.split.client.impressions.ImpressionObserver;
 import io.split.client.impressions.ImpressionsManager;
+import io.split.client.impressions.ImpressionsManagerConfig;
 import io.split.client.impressions.ImpressionsManagerImpl;
+import io.split.client.impressions.ImpressionsTelemetryRecorder;
 import io.split.client.impressions.ImpressionsSender;
+import io.split.client.impressions.TelemetryRuntimeImpressionsRecorder;
+import io.split.client.impressions.TelemetryUniqueKeysSender;
 import io.split.client.impressions.ImpressionsStorage;
 import io.split.client.impressions.ImpressionsStorageConsumer;
 import io.split.client.impressions.ImpressionsStorageProducer;
@@ -137,7 +141,7 @@
 import io.split.client.utils.SplitExecutorFactory;
 import static io.split.client.utils.SplitExecutorFactory.buildExecutorService;
 
 public class SplitFactoryImpl implements SplitFactory {
     private static final org.slf4j.Logger _log = LoggerFactory.getLogger(SplitFactoryImpl.class);
     private static final String LEGACY_LOG_MESSAGE = "The sdk initialize in localhost mode using Legacy file. The splitFile or "
             +
@@ -752,11 +756,12 @@
                 : null;
         ProcessImpressionNone processImpressionNone = new ProcessImpressionNone(listener != null, _uniqueKeysTracker, counter);
 
+        ImpressionsTelemetryRecorder telemetryRecorder = new TelemetryRuntimeImpressionsRecorder(_telemetryStorageProducer);
         switch (config.impressionsMode()) {
             case OPTIMIZED:
                 ImpressionObserver impressionObserver = new ImpressionObserver(config.getLastSeenCacheSize());
                 processImpressionStrategy = new ProcessImpressionOptimized(listener != null, impressionObserver,
-                        counter, _telemetryStorageProducer);
+                        counter, telemetryRecorder);
                 break;
             case DEBUG:
                 impressionObserver = new ImpressionObserver(config.getLastSeenCacheSize());
@@ -766,7 +771,13 @@
                 processImpressionStrategy = processImpressionNone;
                 break;
         }
-        return ImpressionsManagerImpl.instance(config, _telemetryStorageProducer, impressionsStorageConsumer,
+        ImpressionsManagerConfig impressionsManagerConfig = ImpressionsManagerConfig.builder()
+                .mode(config.impressionsMode())
+                .impressionsRefreshRateSeconds(config.impressionsRefreshRate())
+                .threadFactory(config.getThreadFactory())
+                .debugEnabled(config.debugEnabled())
+                .build();
+        return ImpressionsManagerImpl.instance(impressionsManagerConfig, telemetryRecorder, impressionsStorageConsumer,
                 impressionsStorageProducer,
                 _impressionsSender, processImpressionNone, processImpressionStrategy, counter, listener);
     }
@@ -809,7 +820,7 @@
         int uniqueKeysRefreshRate = config.operationMode().equals(OperationMode.STANDALONE)
                 ? config.uniqueKeysRefreshRateInMemory()
                 : config.uniqueKeysRefreshRateRedis();
-        return new UniqueKeysTrackerImp(_telemetrySynchronizer, uniqueKeysRefreshRate,
+        return new UniqueKeysTrackerImp(new TelemetryUniqueKeysSender(_telemetrySynchronizer), uniqueKeysRefreshRate,
                 config.filterUniqueKeysRefreshRate(),
                 config.getThreadFactory());
     }

client/src/main/java/io/split/client/impressions/TelemetryRuntimeImpressionsRecorder.java

@@ -0,0 +1,23 @@
+package io.split.client.impressions;
+
+import io.split.telemetry.domain.enums.ImpressionsDataTypeEnum;
+import io.split.telemetry.storage.TelemetryRuntimeProducer;
+
+public class TelemetryRuntimeImpressionsRecorder implements ImpressionsTelemetryRecorder {
+
+    private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
+
+    public TelemetryRuntimeImpressionsRecorder(TelemetryRuntimeProducer telemetryRuntimeProducer) {
+        _telemetryRuntimeProducer = telemetryRuntimeProducer;
+    }
+
+    @Override
+    public void recordImpressionsDropped(long count) {
+        _telemetryRuntimeProducer.recordImpressionStats(ImpressionsDataTypeEnum.IMPRESSIONS_DROPPED, count);
+    }
+
+    @Override
+    public void recordImpressionsQueued(long count) {
+        _telemetryRuntimeProducer.recordImpressionStats(ImpressionsDataTypeEnum.IMPRESSIONS_QUEUED, count);
+    }
+}

client/src/main/java/io/split/client/impressions/TelemetryUniqueKeysSender.java

@@ -0,0 +1,18 @@
+package io.split.client.impressions;
+
+import io.split.client.dtos.UniqueKeys;
+import io.split.telemetry.synchronizer.TelemetrySynchronizer;
+
+public class TelemetryUniqueKeysSender implements UniqueKeysSender {
+
+    private final TelemetrySynchronizer _telemetrySynchronizer;
+
+    public TelemetryUniqueKeysSender(TelemetrySynchronizer telemetrySynchronizer) {
+        _telemetrySynchronizer = telemetrySynchronizer;
+    }
+
+    @Override
+    public void send(UniqueKeys uniqueKeys) {
+        _telemetrySynchronizer.synchronizeUniqueKeys(uniqueKeys);
+    }
+}

impressions/pom.xml

@@ -0,0 +1,65 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+         http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>io.split.client</groupId>
+    <artifactId>java-client-parent</artifactId>
+    <version>4.18.3</version>
+  </parent>
+
+  <artifactId>impressions</artifactId>
+  <packaging>jar</packaging>
+  <name>Impressions</name>
+  <description>Impression tracking core: dedup strategies, bloom-filter unique-key tracking, counter aggregation</description>
+
+  <dependencies>
+    <dependency>
+      <groupId>com.google.code.gson</groupId>
+      <artifactId>gson</artifactId>
+      <version>2.13.1</version>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+      <version>1.7.36</version>
+    </dependency>
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+      <version>32.0.1-jre</version>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-all</artifactId>
+      <version>1.3</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <version>1.10.19</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.powermock</groupId>
+      <artifactId>powermock-module-junit4</artifactId>
+      <version>1.7.4</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.powermock</groupId>
+      <artifactId>powermock-api-mockito</artifactId>
+      <version>1.7.4</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>

client/src/main/java/io/split/client/impressions/AsynchronousImpressionListener.java → impressions/src/main/java/io/split/client/impressions/AsynchronousImpressionListener.java

@@ -1,6 +1,5 @@
 package io.split.client.impressions;
 
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -9,6 +8,7 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * A wrapper around an ImpressionListener provided by the customer. The purpose
@@ -25,10 +25,12 @@ public class AsynchronousImpressionListener implements ImpressionListener {
     private final ExecutorService _executor;
 
     public static AsynchronousImpressionListener build(ImpressionListener delegate, int capacity) {
-        ThreadFactory threadFactory = new ThreadFactoryBuilder()
-                .setDaemon(true)
-                .setNameFormat("impression-listener-wrapper-%d")
-                .build();
+        AtomicInteger counter = new AtomicInteger(0);
+        ThreadFactory threadFactory = r -> {
+            Thread t = new Thread(r, "impression-listener-wrapper-" + counter.getAndIncrement());
+            t.setDaemon(true);
+            return t;
+        };
 
         ExecutorService executor = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(capacity), threadFactory);
 

client/src/main/java/io/split/client/impressions/ImpressionCounter.java → impressions/src/main/java/io/split/client/impressions/ImpressionCounter.java

@@ -5,16 +5,14 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-
 public class ImpressionCounter {
 
     public static class Key {
         private final String _featureName;
         private final long _timeFrame;
 
         public Key(String featureFlagName, long timeframe) {
-            _featureName = checkNotNull(featureFlagName);
+            _featureName = Objects.requireNonNull(featureFlagName);
             _timeFrame = timeframe;
         }
 

impressions/src/main/java/io/split/client/impressions/ImpressionsManagerConfig.java

@@ -0,0 +1,39 @@
+package io.split.client.impressions;
+
+import java.util.concurrent.ThreadFactory;
+
+public final class ImpressionsManagerConfig {
+
+    private final ImpressionsManager.Mode _mode;
+    private final int _impressionsRefreshRateSeconds;
+    private final ThreadFactory _threadFactory;
+    private final boolean _debugEnabled;
+
+    private ImpressionsManagerConfig(Builder builder) {
+        _mode = builder._mode;
+        _impressionsRefreshRateSeconds = builder._impressionsRefreshRateSeconds;
+        _threadFactory = builder._threadFactory;
+        _debugEnabled = builder._debugEnabled;
+    }
+
+    public ImpressionsManager.Mode mode() { return _mode; }
+    public int impressionsRefreshRateSeconds() { return _impressionsRefreshRateSeconds; }
+    public ThreadFactory threadFactory() { return _threadFactory; }
+    public boolean debugEnabled() { return _debugEnabled; }
+
+    public static Builder builder() { return new Builder(); }
+
+    public static final class Builder {
+        private ImpressionsManager.Mode _mode = ImpressionsManager.Mode.OPTIMIZED;
+        private int _impressionsRefreshRateSeconds = 60;
+        private ThreadFactory _threadFactory = null;
+        private boolean _debugEnabled = false;
+
+        public Builder mode(ImpressionsManager.Mode mode) { _mode = mode; return this; }
+        public Builder impressionsRefreshRateSeconds(int rate) { _impressionsRefreshRateSeconds = rate; return this; }
+        public Builder threadFactory(ThreadFactory tf) { _threadFactory = tf; return this; }
+        public Builder debugEnabled(boolean debug) { _debugEnabled = debug; return this; }
+
+        public ImpressionsManagerConfig build() { return new ImpressionsManagerConfig(this); }
+    }
+}