Release 4.9.0

CHANGES.txt

@@ -1,3 +1,8 @@
+4.9.0 (Sep 8, 2023)
+- Added InputStream config for localhost mode providing a solution when the file is inside a jar.
+- Fixed track impressions to send all impressions to the listener.
+- Fixed SyncManager shutdown to stop SSE only when is streaming mode on.
+
 4.8.1 (Aug 1, 2023)
 - Applied linting rules to the code.
 - Fixed an issue when the prefix is empty for Redis settings.

client/pom.xml

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

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

@@ -2,55 +2,44 @@
 
 import com.google.gson.stream.JsonReader;
 import io.split.client.dtos.SplitChange;
+import io.split.client.utils.InputStreamProvider;
 import io.split.client.utils.Json;
 import io.split.client.utils.LocalhostSanitizer;
 import io.split.engine.common.FetchOptions;
 import io.split.engine.experiments.SplitChangeFetcher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileReader;
-import java.io.UnsupportedEncodingException;
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.util.Arrays;
 
 public class JsonLocalhostSplitChangeFetcher implements SplitChangeFetcher {
 
     private static final Logger _log = LoggerFactory.getLogger(JsonLocalhostSplitChangeFetcher.class);
-    private final File _file;
+    private final InputStreamProvider _inputStreamProvider;
     private byte [] lastHash;
 
-    public JsonLocalhostSplitChangeFetcher(String filePath) {
-        _file = new File(filePath);
+    public JsonLocalhostSplitChangeFetcher(InputStreamProvider inputStreamProvider) {
+        _inputStreamProvider = inputStreamProvider;
         lastHash = new byte[0];
     }
 
     @Override
     public SplitChange fetch(long since, FetchOptions options) {
-
         try {
-            JsonReader jsonReader = new JsonReader(new FileReader(_file));
+            JsonReader jsonReader = new JsonReader(new BufferedReader(new InputStreamReader(_inputStreamProvider.get(), StandardCharsets.UTF_8)));
             SplitChange splitChange = Json.fromJson(jsonReader, SplitChange.class);
             return processSplitChange(splitChange, since);
-        } catch (FileNotFoundException f){
-            _log.warn(String.format("There was no file named %s found. " +
-                            "We created a split client that returns default treatments for all feature flags for all of your users. " +
-                            "If you wish to return a specific treatment for a feature flag, enter the name of that feature flag name and " +
-                            "treatment name separated by whitespace in %s; one pair per line. Empty lines or lines starting with '#' are " +
-                            "considered comments",
-                    _file.getPath(), _file.getPath()), f);
-            throw new IllegalStateException("Problem fetching splitChanges: " + f.getMessage(), f);
         } catch (Exception e) {
-            _log.warn(String.format("Problem to fetch split change using the file %s",
-                    _file.getPath()), e);
             throw new IllegalStateException("Problem fetching splitChanges: " + e.getMessage(), e);
         }
     }
 
-    private SplitChange processSplitChange(SplitChange splitChange, long changeNumber) throws NoSuchAlgorithmException, UnsupportedEncodingException {
+    private SplitChange processSplitChange(SplitChange splitChange, long changeNumber) throws NoSuchAlgorithmException {
         SplitChange splitChangeToProcess = LocalhostSanitizer.sanitization(splitChange);
         // if the till is less than storage CN and different from the default till ignore the change
         if (splitChangeToProcess.till < changeNumber && splitChangeToProcess.till != -1) {

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

@@ -2,13 +2,15 @@
 
 import io.split.client.impressions.ImpressionListener;
 import io.split.client.impressions.ImpressionsManager;
+import io.split.client.utils.FileTypeEnum;
 import io.split.integrations.IntegrationsConfig;
 import io.split.storages.enums.OperationMode;
 import io.split.storages.enums.StorageMode;
 import org.apache.hc.core5.http.HttpHost;
 import pluggable.CustomStorageWrapper;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.Properties;
 import java.util.concurrent.ThreadFactory;
 
@@ -49,6 +51,8 @@ public class SplitClientConfig {
     private final int _maxStringLength;
     private final boolean _destroyOnShutDown;
     private final String _splitFile;
+    private final FileTypeEnum _fileType;
+    private final InputStream _inputStream;
     private final String _segmentDirectory;
     private final IntegrationsConfig _integrationsConfig;
     private final boolean _streamingEnabled;
@@ -81,7 +85,6 @@ public class SplitClientConfig {
     public static String splitSdkVersion;
     private final long _lastSeenCacheSize;
 
-
     public static Builder builder() {
         return new Builder();
     }
@@ -111,6 +114,8 @@ private SplitClientConfig(String endpoint,
                               int maxStringLength,
                               boolean destroyOnShutDown,
                               String splitFile,
+                              FileTypeEnum fileType,
+                              InputStream inputStream,
                               String segmentDirectory,
                               IntegrationsConfig integrationsConfig,
                               boolean streamingEnabled,
@@ -159,6 +164,8 @@ private SplitClientConfig(String endpoint,
         _maxStringLength = maxStringLength;
         _destroyOnShutDown = destroyOnShutDown;
         _splitFile = splitFile;
+        _fileType = fileType;
+        _inputStream = inputStream;
         _segmentDirectory = segmentDirectory;
         _integrationsConfig = integrationsConfig;
         _streamingEnabled = streamingEnabled;
@@ -301,6 +308,14 @@ public String splitFile() {
         return _splitFile;
     }
 
+    public FileTypeEnum fileType() {
+        return _fileType;
+    }
+
+    public InputStream inputStream(){
+        return _inputStream;
+    }
+
     public String segmentDirectory() {
         return _segmentDirectory;
     }
@@ -394,6 +409,8 @@ public static final class Builder {
         private int _maxStringLength = 250;
         private boolean _destroyOnShutDown = true;
         private String _splitFile = null;
+        private FileTypeEnum _fileType = null;
+        private InputStream _inputStream = null;
         private String _segmentDirectory = null;
         private IntegrationsConfig _integrationsConfig = null;
         private boolean _streamingEnabled = true;
@@ -748,6 +765,12 @@ public Builder splitFile(String splitFile) {
             return this;
         }
 
+        public Builder splitFile(InputStream inputStream, FileTypeEnum fileType) {
+            _fileType = fileType;
+            _inputStream = inputStream;
+            return this;
+        }
+
         /**
          * Set the location of the directory where are the segment json files for localhost mode.
          * This setting is optional.
@@ -1005,6 +1028,8 @@ public SplitClientConfig build() {
                     _maxStringLength,
                     _destroyOnShutDown,
                     _splitFile,
+                    _fileType,
+                    _inputStream,
                     _segmentDirectory,
                     _integrationsConfig,
                     _streamingEnabled,

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

@@ -1,5 +1,6 @@
 package io.split.client;
 
+import com.google.common.io.Files;
 import io.split.client.dtos.Metadata;
 import io.split.client.events.EventsSender;
 import io.split.client.events.EventsStorage;
@@ -30,7 +31,11 @@
 import io.split.client.interceptors.GzipDecoderResponseInterceptor;
 import io.split.client.interceptors.GzipEncoderRequestInterceptor;
 import io.split.client.interceptors.SdkMetadataInterceptorFilter;
+import io.split.client.utils.FileInputStreamProvider;
+import io.split.client.utils.FileTypeEnum;
+import io.split.client.utils.InputStreamProvider;
 import io.split.client.utils.SDKMetadata;
+import io.split.client.utils.StaticContentInputStreamProvider;
 import io.split.engine.SDKReadinessGates;
 import io.split.engine.common.ConsumerSyncManager;
 import io.split.engine.common.ConsumerSynchronizer;
@@ -99,6 +104,7 @@
 import pluggable.CustomStorageWrapper;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.net.InetAddress;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -111,6 +117,8 @@
 
 public class SplitFactoryImpl implements SplitFactory {
     private static final Logger _log = LoggerFactory.getLogger(SplitFactory.class);
+    private static final String LEGACY_LOG_MESSAGE = "The sdk initialize in localhost mode using Legacy file. The splitFile or " +
+                                                     "inputStream doesn't add it to the config.";
     private final static long SSE_CONNECT_TIMEOUT = 30000;
     private final static long SSE_SOCKET_TIMEOUT = 70000;
 
@@ -366,16 +374,7 @@ protected SplitFactoryImpl(SplitClientConfig config) {
                config.getThreadFactory());
 
         // SplitFetcher
-        SplitChangeFetcher splitChangeFetcher;
-        String splitFile = config.splitFile();
-        if (splitFile != null && splitFile.toLowerCase().endsWith(".json")){
-            splitChangeFetcher = new JsonLocalhostSplitChangeFetcher(config.splitFile());
-        } else if (splitFile != null && !splitFile.isEmpty() && (splitFile.endsWith(".yaml") || splitFile.endsWith(".yml"))) {
-            splitChangeFetcher = new YamlLocalhostSplitChangeFetcher(splitFile);
-        } else {
-            splitChangeFetcher = new LegacyLocalhostSplitChangeFetcher(config.splitFile());
-        }
-
+        SplitChangeFetcher splitChangeFetcher = createSplitChangeFetcher(config);
         SplitParser splitParser = new SplitParser();
 
         _splitFetcher = new SplitFetcherImp(splitChangeFetcher, splitParser, splitCache, _telemetryStorageProducer);
@@ -643,4 +642,53 @@ private UniqueKeysTracker createUniqueKeysTracker(SplitClientConfig config){
         }
         return null;
     }
+
+    private SplitChangeFetcher createSplitChangeFetcher(SplitClientConfig splitClientConfig) {
+        String splitFile = splitClientConfig.splitFile();
+        InputStream inputStream = splitClientConfig.inputStream();
+        FileTypeEnum fileType = splitClientConfig.fileType();
+        InputStreamProvider inputStreamProvider;
+        if (splitFile != null || !isInputStreamConfigValid(inputStream, fileType)) {
+            if (splitFile == null) {
+                _log.warn("The InputStream config is invalid");
+            }
+            fileType = getFileTypeFromFileName(splitFile);
+            inputStreamProvider = new FileInputStreamProvider(splitFile);
+        } else {
+            inputStreamProvider = new StaticContentInputStreamProvider(inputStream);
+        }
+        try {
+            switch (fileType){
+                case JSON:
+                    return new JsonLocalhostSplitChangeFetcher(inputStreamProvider);
+                case YAML:
+                    return new YamlLocalhostSplitChangeFetcher(inputStreamProvider);
+                default:
+                    _log.warn(LEGACY_LOG_MESSAGE);
+                    return new LegacyLocalhostSplitChangeFetcher(splitFile);
+            }
+        } catch (Exception e) {
+            _log.warn(String.format("There was no file named %s found. " +
+                            "We created a split client that returns default treatments for all feature flags for all of your users. " +
+                            "If you wish to return a specific treatment for a feature flag, enter the name of that feature flag name and " +
+                            "treatment name separated by whitespace in %s; one pair per line. Empty lines or lines starting with '#' are " +
+                            "considered comments",
+                    splitFile, splitFile), e);
+        }
+        _log.warn(LEGACY_LOG_MESSAGE);
+        return new LegacyLocalhostSplitChangeFetcher(splitFile);
+    }
+
+    private Boolean isInputStreamConfigValid(InputStream inputStream, FileTypeEnum fileType) {
+        return inputStream != null && fileType != null;
+    }
+
+    private FileTypeEnum getFileTypeFromFileName(String fileName) {
+        try {
+            return FileTypeEnum.valueOf(Files.getFileExtension(fileName).toUpperCase());
+        } catch (Exception e) {
+            return FileTypeEnum.LEGACY;
+        }
+
+    }
 }

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

@@ -5,16 +5,15 @@
 import io.split.client.dtos.Split;
 import io.split.client.dtos.SplitChange;
 import io.split.client.dtos.Status;
+import io.split.client.utils.InputStreamProvider;
 import io.split.client.utils.LocalhostConstants;
 import io.split.engine.common.FetchOptions;
 import io.split.engine.experiments.SplitChangeFetcher;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.yaml.snakeyaml.Yaml;
 
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileReader;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -23,21 +22,20 @@
 
 import static io.split.client.utils.LocalhostSanitizer.createCondition;
 
-
 public class YamlLocalhostSplitChangeFetcher implements SplitChangeFetcher {
 
     private static final Logger _log = LoggerFactory.getLogger(YamlLocalhostSplitChangeFetcher.class);
-    private final File _splitFile;
+    private final InputStreamProvider _inputStreamProvider;
 
-    public YamlLocalhostSplitChangeFetcher(String filePath) {
-        _splitFile = new File(filePath);
+    public YamlLocalhostSplitChangeFetcher(InputStreamProvider inputStreamProvider) {
+        _inputStreamProvider = inputStreamProvider;
     }
 
     @Override
     public SplitChange fetch(long since, FetchOptions options) {
         try {
             Yaml yaml = new Yaml();
-            List<Map<String, Map<String, Object>>> yamlSplits = yaml.load(new FileReader(_splitFile));
+            List<Map<String, Map<String, Object>>> yamlSplits = yaml.load(_inputStreamProvider.get());
             SplitChange splitChange = new SplitChange();
             splitChange.splits = new ArrayList<>();
             for(Map<String, Map<String, Object>> aSplit : yamlSplits) {
@@ -76,17 +74,8 @@ public SplitChange fetch(long since, FetchOptions options) {
             splitChange.till = since;
             splitChange.since = since;
             return splitChange;
-        } catch (FileNotFoundException f) {
-            _log.warn(String.format("There was no file named %s found. We created a split client that returns default treatments " +
-                            "for all feature flags for all of your users. If you wish to return a specific treatment for a feature flag, " +
-                            "enter the name of that feature flag name and treatment name separated by whitespace in %s; one pair per line. " +
-                            "Empty lines or lines starting with '#' are considered comments",
-                    _splitFile.getPath(), _splitFile.getPath()), f);
-            throw new IllegalStateException("Problem fetching splitChanges: " + f.getMessage(), f);
         } catch (Exception e) {
-            _log.warn(String.format("Problem to fetch split change using the file %s",
-                    _splitFile.getPath()), e);
-            throw new IllegalStateException("Problem fetching splitChanges: " + e.getMessage(), e);
+            throw new IllegalStateException("Problem fetching splitChanges using a yaml file: " + e.getMessage(), e);
         }
     }
 }

client/src/main/java/io/split/client/exceptions/InputStreamProviderException.java

@@ -0,0 +1,8 @@
+package io.split.client.exceptions;
+
+public class InputStreamProviderException extends Exception {
+
+    public InputStreamProviderException(String message) {
+        super(message);
+    }
+}

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

@@ -61,7 +61,7 @@ public static ImpressionsManagerImpl instanceForTest(SplitClientConfig config,
                                                          ImpressionsStorageProducer impressionsStorageProducer,
                                                          ProcessImpressionStrategy processImpressionStrategy,
                                                          ImpressionCounter counter,
-                                                         ImpressionListener listener) throws URISyntaxException {
+                                                         ImpressionListener listener) {
         return new ImpressionsManagerImpl(config, impressionsSender, telemetryRuntimeProducer, impressionsStorageConsumer,
                 impressionsStorageProducer, processImpressionStrategy, counter, listener);
     }
@@ -73,7 +73,7 @@ private ImpressionsManagerImpl(SplitClientConfig config,
                                    ImpressionsStorageProducer impressionsStorageProducer,
                                    ProcessImpressionStrategy processImpressionStrategy,
                                    ImpressionCounter impressionCounter,
-                                   ImpressionListener impressionListener) throws URISyntaxException {
+                                   ImpressionListener impressionListener) {
 
 
         _config = checkNotNull(config);
@@ -117,7 +117,7 @@ public void track(List<Impression> impressions) {
 
         ImpressionsResult impressionsResult = _processImpressionStrategy.process(impressions);
         List<Impression> impressionsForLogs = impressionsResult.getImpressionsToQueue();
-        List<Impression> impressionsToListener = impressionsResult.getImpressionsToQueue();
+        List<Impression> impressionsToListener = impressionsResult.getImpressionsToListener();
 
         int totalImpressions = impressionsForLogs.size();
         long queued = _impressionsStorageProducer.put(impressionsForLogs.stream().map(KeyImpression::fromImpression).collect(Collectors.toList()));