Task/streaming decorator

client/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.split.client</groupId>
         <artifactId>java-client-parent</artifactId>
-        <version>4.12.0-rc1</version>
+        <version>4.12.0-rc2</version>
     </parent>
     <artifactId>java-client</artifactId>
     <packaging>jar</packaging>

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

@@ -0,0 +1,18 @@
+package io.split.client;
+
+import io.split.client.CustomHeaderDecorator;
+import io.split.client.dtos.RequestContext;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+class NoOpHeaderDecorator implements CustomHeaderDecorator {
+    public NoOpHeaderDecorator() {
+    }
+
+    @Override
+    public Map<String, List<String>> getHeaderOverrides(RequestContext context) {
+        return new HashMap<>();
+    }
+}

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

@@ -2,7 +2,6 @@
 
 import io.split.client.dtos.RequestContext;
 
-//`import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase;
 import org.apache.hc.core5.http.HttpRequest;
 import org.apache.hc.core5.http.Header;
 
@@ -14,16 +13,6 @@
 import java.util.Set;
 import java.util.List;
 
-class NoOpHeaderDecorator implements CustomHeaderDecorator {
-    public NoOpHeaderDecorator() {
-    }
-
-    @Override
-    public Map<String, List<String>> getHeaderOverrides(RequestContext context) {
-        return new HashMap<>();
-    }
-}
-
 public final class RequestDecorator {
     CustomHeaderDecorator _headerDecorator;
 

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

@@ -187,7 +187,8 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
         _gates = new SDKReadinessGates();
 
         // HttpClient
-        _splitHttpClient = buildSplitHttpClient(apiToken, config, _sdkMetadata);
+        RequestDecorator requestDecorator = new RequestDecorator(config.customHeaderDecorator());
+        _splitHttpClient = buildSplitHttpClient(apiToken, config, _sdkMetadata, requestDecorator);
 
         // Roots
         _rootTarget = URI.create(config.endpoint());
@@ -258,7 +259,7 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
         // SyncManager
         SplitTasks splitTasks = SplitTasks.build(_splitSynchronizationTask, _segmentSynchronizationTaskImp,
                 _impressionsManager, _eventsTask, _telemetrySyncTask, _uniqueKeysTracker);
-        SplitAPI splitAPI = SplitAPI.build(_splitHttpClient, buildSSEdHttpClient(apiToken, config, _sdkMetadata));
+        SplitAPI splitAPI = SplitAPI.build(_splitHttpClient, buildSSEdHttpClient(apiToken, config, _sdkMetadata), requestDecorator);
 
         _syncManager = SyncManagerImp.build(splitTasks, _splitFetcher, splitCache, splitAPI,
                 segmentCache, _gates, _telemetryStorageProducer, _telemetrySynchronizer, config, splitParser,
@@ -495,7 +496,7 @@ public boolean isDestroyed() {
     }
 
     private static SplitHttpClient buildSplitHttpClient(String apiToken, SplitClientConfig config,
-            SDKMetadata sdkMetadata)
+            SDKMetadata sdkMetadata, RequestDecorator requestDecorator)
             throws URISyntaxException {
         SSLConnectionSocketFactory sslSocketFactory = SSLConnectionSocketFactoryBuilder.create()
                 .setSslContext(SSLContexts.createSystemDefault())
@@ -529,7 +530,7 @@ private static SplitHttpClient buildSplitHttpClient(String apiToken, SplitClient
         }
 
         return SplitHttpClientImpl.create(httpClientbuilder.build(),
-                new RequestDecorator(config.customHeaderDecorator()),
+                requestDecorator,
                 apiToken,
                 sdkMetadata);
     }

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

@@ -83,7 +83,7 @@ public static PushManagerImp build(Synchronizer synchronizer,
         PushStatusTracker pushStatusTracker = new PushStatusTrackerImp(statusMessages, telemetryRuntimeProducer);
         return new PushManagerImp(new AuthApiClientImp(authUrl, splitAPI.getHttpClient(), telemetryRuntimeProducer),
                 EventSourceClientImp.build(streamingUrl, featureFlagsWorker, segmentWorker, pushStatusTracker, splitAPI.getSseHttpClient(),
-                        telemetryRuntimeProducer, threadFactory),
+                        telemetryRuntimeProducer, threadFactory, splitAPI.getRequestDecorator()),
                 featureFlagsWorker,
                 segmentWorker,
                 pushStatusTracker,

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

@@ -1,5 +1,6 @@
 package io.split.engine.common;
 
+import io.split.client.RequestDecorator;
 import io.split.service.SplitHttpClient;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.slf4j.Logger;
@@ -9,15 +10,17 @@ public class SplitAPI {
 
     private final SplitHttpClient _httpClient;
     private final CloseableHttpClient _sseHttpClient;
+    private final RequestDecorator _requestDecorator;
     private static final Logger _log = LoggerFactory.getLogger(SplitAPI.class);
 
-    private SplitAPI(SplitHttpClient httpClient, CloseableHttpClient sseHttpClient) {
+    private SplitAPI(SplitHttpClient httpClient, CloseableHttpClient sseHttpClient, RequestDecorator requestDecorator) {
         _httpClient = httpClient;
         _sseHttpClient = sseHttpClient;
+        _requestDecorator = requestDecorator;
     }
 
-    public static SplitAPI build(SplitHttpClient httpClient, CloseableHttpClient sseHttpClient){
-        return new SplitAPI(httpClient,sseHttpClient);
+    public static SplitAPI build(SplitHttpClient httpClient, CloseableHttpClient sseHttpClient, RequestDecorator requestDecorator){
+        return new SplitAPI(httpClient, sseHttpClient, requestDecorator);
     }
 
     public SplitHttpClient getHttpClient() {
@@ -28,6 +31,8 @@ public CloseableHttpClient getSseHttpClient() {
         return _sseHttpClient;
     }
 
+    public RequestDecorator getRequestDecorator() { return  _requestDecorator; }
+
     public void close(){
         try {
             _sseHttpClient.close();

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

@@ -2,6 +2,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Strings;
+import io.split.client.RequestDecorator;
 import io.split.engine.sse.client.RawEvent;
 import io.split.engine.sse.client.SSEClient;
 import io.split.engine.sse.dtos.SegmentQueueDto;
@@ -40,7 +41,8 @@ public class EventSourceClientImp implements EventSourceClient {
                                                PushStatusTracker pushStatusTracker,
                                                CloseableHttpClient sseHttpClient,
                                                TelemetryRuntimeProducer telemetryRuntimeProducer,
-                                               ThreadFactory threadFactory) {
+                                               ThreadFactory threadFactory,
+                                               RequestDecorator requestDecorator) {
         _baseStreamingUrl = checkNotNull(baseStreamingUrl);
         _notificationParser = checkNotNull(notificationParser);
         _notificationProcessor = checkNotNull(notificationProcessor);
@@ -51,7 +53,8 @@ public class EventSourceClientImp implements EventSourceClient {
                 status -> { _pushStatusTracker.handleSseStatus(status); return null; },
                 sseHttpClient,
                 telemetryRuntimeProducer,
-                threadFactory);
+                threadFactory,
+                requestDecorator);
         _firstEvent = new AtomicBoolean();
     }
 
@@ -61,14 +64,16 @@ public static EventSourceClientImp build(String baseStreamingUrl,
                                              PushStatusTracker pushStatusTracker,
                                              CloseableHttpClient sseHttpClient,
                                              TelemetryRuntimeProducer telemetryRuntimeProducer,
-                                             ThreadFactory threadFactory) {
+                                             ThreadFactory threadFactory,
+                                             RequestDecorator requestDecorator) {
         return new EventSourceClientImp(baseStreamingUrl,
                 new NotificationParserImp(),
                 NotificationProcessorImp.build(featureFlagsWorker, segmentWorker, pushStatusTracker),
                 pushStatusTracker,
                 sseHttpClient,
                 telemetryRuntimeProducer,
-                threadFactory);
+                threadFactory,
+                requestDecorator);
     }
 
     @Override

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

@@ -1,10 +1,12 @@
 package io.split.engine.sse.client;
 
 import com.google.common.base.Strings;
+import io.split.client.RequestDecorator;
 import io.split.telemetry.domain.StreamingEvent;
 import io.split.telemetry.domain.enums.StreamEventsEnum;
 import io.split.telemetry.storage.TelemetryRuntimeProducer;
 import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.classic.methods.HttpPost;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
 import org.slf4j.Logger;
@@ -56,20 +58,23 @@ private enum ConnectionState {
     private final AtomicReference<CloseableHttpResponse> _ongoingResponse = new AtomicReference<>();
     private final AtomicReference<HttpGet> _ongoingRequest = new AtomicReference<>();
     private AtomicBoolean _forcedStop;
+    private final RequestDecorator _requestDecorator;
 
     private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
 
     public SSEClient(Function<RawEvent, Void> eventCallback,
                      Function<StatusMessage, Void> statusCallback,
                      CloseableHttpClient client,
                      TelemetryRuntimeProducer telemetryRuntimeProducer,
-                     ThreadFactory threadFactory) {
+                     ThreadFactory threadFactory,
+                     RequestDecorator requestDecorator) {
         _eventCallback = eventCallback;
         _statusCallback = statusCallback;
         _client = client;
         _forcedStop = new AtomicBoolean();
         _telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
         _connectionExecutor = buildExecutorService(threadFactory, "SPLIT-SSEConnection-%d");
+        _requestDecorator = requestDecorator;
     }
 
     public synchronized boolean open(URI uri) {
@@ -177,7 +182,9 @@ private void connectAndLoop(URI uri, CountDownLatch signal) {
     }
 
     private boolean establishConnection(URI uri, CountDownLatch signal) {
-        _ongoingRequest.set(new HttpGet(uri));
+        HttpGet request = new HttpGet(uri);
+        request = (HttpGet) _requestDecorator.decorateHeaders(request);
+        _ongoingRequest.set(request);
         try {
             _ongoingResponse.set(_client.execute(_ongoingRequest.get()));
             if (_ongoingResponse.get().getCode() != 200) {

client/src/test/java/io/split/engine/sse/EventSourceClientTest.java

@@ -1,6 +1,7 @@
 package io.split.engine.sse;
 
 import io.split.SSEMockServer;
+import io.split.client.RequestDecorator;
 import io.split.engine.sse.client.SSEClient;
 import io.split.engine.sse.dtos.ErrorNotification;
 import io.split.engine.sse.dtos.FeatureFlagChangeNotification;
@@ -42,7 +43,7 @@ public void startShouldConnect() throws IOException {
         TelemetryRuntimeProducer telemetryRuntimeProducer = Mockito.mock(InMemoryTelemetryStorage.class);
         sseServer.start();
 
-        EventSourceClient eventSourceClient = new EventSourceClientImp("http://localhost:" + sseServer.getPort(), _notificationParser, _notificationProcessor, _pushStatusTracker, buildHttpClient(), telemetryRuntimeProducer, null);
+        EventSourceClient eventSourceClient = new EventSourceClientImp("http://localhost:" + sseServer.getPort(), _notificationParser, _notificationProcessor, _pushStatusTracker, buildHttpClient(), telemetryRuntimeProducer, null, new RequestDecorator(null));
 
         boolean result = eventSourceClient.start("channel-test", "token-test");
 
@@ -57,7 +58,7 @@ public void startShouldReconnect() throws IOException {
         SSEMockServer sseServer = buildSSEMockServer(eventQueue);
         TelemetryRuntimeProducer telemetryRuntimeProducer = Mockito.mock(InMemoryTelemetryStorage.class);
         sseServer.start();
-        EventSourceClient eventSourceClient = new EventSourceClientImp("http://fake:" + sseServer.getPort(), _notificationParser, _notificationProcessor, _pushStatusTracker, buildHttpClient(), telemetryRuntimeProducer, null);
+        EventSourceClient eventSourceClient = new EventSourceClientImp("http://fake:" + sseServer.getPort(), _notificationParser, _notificationProcessor, _pushStatusTracker, buildHttpClient(), telemetryRuntimeProducer, null, new RequestDecorator(null));
 
         boolean result = eventSourceClient.start("channel-test", "token-test");
 
@@ -74,7 +75,7 @@ public void startAndReceiveNotification() throws IOException {
         SSEMockServer sseServer = buildSSEMockServer(eventQueue);
         TelemetryRuntimeProducer telemetryRuntimeProducer = Mockito.mock(InMemoryTelemetryStorage.class);
         sseServer.start();
-        EventSourceClient eventSourceClient = new EventSourceClientImp("http://localhost:" + sseServer.getPort(), _notificationParser, _notificationProcessor, _pushStatusTracker, buildHttpClient(), telemetryRuntimeProducer, null);
+        EventSourceClient eventSourceClient = new EventSourceClientImp("http://localhost:" + sseServer.getPort(), _notificationParser, _notificationProcessor, _pushStatusTracker, buildHttpClient(), telemetryRuntimeProducer, null, new RequestDecorator(null));
 
         boolean result = eventSourceClient.start("channel-test", "token-test");
 

client/src/test/java/io/split/engine/sse/SSEClientTest.java

@@ -1,5 +1,6 @@
 package io.split.engine.sse;
 
+import io.split.client.RequestDecorator;
 import io.split.engine.sse.client.SSEClient;
 import io.split.telemetry.storage.InMemoryTelemetryStorage;
 import io.split.telemetry.storage.TelemetryRuntimeProducer;
@@ -38,7 +39,7 @@ public void basicUsageTest() throws URISyntaxException, InterruptedException {
         CloseableHttpClient httpClient =  httpClientbuilder.build();
 
         SSEClient sse = new SSEClient(e -> null,
-                s -> null, httpClient, telemetryRuntimeProducer, null);
+                s -> null, httpClient, telemetryRuntimeProducer, null, new RequestDecorator(null));
         sse.open(uri);
         Thread.sleep(5000);
         sse.close();

pluggable-storage/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <artifactId>java-client-parent</artifactId>
         <groupId>io.split.client</groupId>
-        <version>4.12.0-rc1</version>
+        <version>4.12.0-rc2</version>
     </parent>
 
     <version>2.1.0</version>

pom.xml

@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>io.split.client</groupId>
     <artifactId>java-client-parent</artifactId>
-    <version>4.12.0-rc1</version>
+    <version>4.12.0-rc2</version>
     <dependencyManagement>
         <dependencies>
             <dependency>

redis-wrapper/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <artifactId>java-client-parent</artifactId>
         <groupId>io.split.client</groupId>
-        <version>4.12.0-rc1</version>
+        <version>4.12.0-rc2</version>
     </parent>
     <artifactId>redis-wrapper</artifactId>
     <version>3.1.0</version>

testing/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.split.client</groupId>
         <artifactId>java-client-parent</artifactId>
-        <version>4.12.0-rc1</version>
+        <version>4.12.0-rc2</version>
     </parent>
     <artifactId>java-client-testing</artifactId>
     <packaging>jar</packaging>