launchdarkly
diff --git a/‎build.gradle‎
Lines changed: 1 addition & 1 deletion b/‎build.gradle‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎contract-tests/service/src/main/java/sdktest/Representations.java‎
Lines changed: 1 addition & 0 deletions b/‎contract-tests/service/src/main/java/sdktest/Representations.java‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎contract-tests/service/src/main/java/sdktest/SdkClientEntity.java‎
Lines changed: 1 addition & 0 deletions b/‎contract-tests/service/src/main/java/sdktest/SdkClientEntity.java‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎contract-tests/service/src/main/java/sdktest/TestService.java‎
Lines changed: 2 additions & 1 deletion b/‎contract-tests/service/src/main/java/sdktest/TestService.java‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎src/main/java/com/launchdarkly/sdk/server/ComponentsImpl.java‎
Lines changed: 20 additions & 24 deletions b/‎src/main/java/com/launchdarkly/sdk/server/ComponentsImpl.java‎
Lines changed: 20 additions & 24 deletions
diff --git a/‎src/main/java/com/launchdarkly/sdk/server/DefaultFeatureRequestor.java‎
Lines changed: 35 additions & 15 deletions b/‎src/main/java/com/launchdarkly/sdk/server/DefaultFeatureRequestor.java‎
Lines changed: 35 additions & 15 deletions
diff --git a/‎src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java‎
Lines changed: 19 additions & 8 deletions b/‎src/main/java/com/launchdarkly/sdk/server/StreamProcessor.java‎
Lines changed: 19 additions & 8 deletions
diff --git a/‎src/main/java/com/launchdarkly/sdk/server/integrations/PollingDataSourceBuilder.java‎
Lines changed: 14 additions & 0 deletions b/‎src/main/java/com/launchdarkly/sdk/server/integrations/PollingDataSourceBuilder.java‎
Lines changed: 14 additions & 0 deletions
diff --git a/‎src/main/java/com/launchdarkly/sdk/server/integrations/StreamingDataSourceBuilder.java‎
Lines changed: 16 additions & 2 deletions b/‎src/main/java/com/launchdarkly/sdk/server/integrations/StreamingDataSourceBuilder.java‎
Lines changed: 16 additions & 2 deletions
@@ -73,7 +73,7 @@ ext.versions = [
     "guava": "30.1-jre",
     "jackson": "2.11.2", 
     "launchdarklyJavaSdkCommon": "2.0.0",
-    "launchdarklyJavaSdkInternal": "1.0.0",
+    "launchdarklyJavaSdkInternal": "1.1.0",
     "launchdarklyLogging": "1.1.0",
     "okhttp": "4.9.3", // specify this for the SDK build instead of relying on the transitive dependency from okhttp-eventsource
     "okhttpEventsource": "4.1.0",
 
@@ -35,6 +35,7 @@ public static class SdkConfigParams {
   public static class SdkConfigStreamParams {
     URI baseUri;
     long initialRetryDelayMs;
+    String filter;
   }
 
   public static class SdkConfigEventParams {
 
@@ -300,6 +300,7 @@ private LDConfig buildSdkConfig(SdkConfigParams params, String tag) {
       if (params.streaming.initialRetryDelayMs > 0) {
         dataSource.initialReconnectDelay(Duration.ofMillis(params.streaming.initialRetryDelayMs));
       }
+      dataSource.payloadFilter(params.streaming.filter);
       builder.dataSource(dataSource);
     }
 
 
@@ -32,7 +32,8 @@ public class TestService {
     "context-type",
     "service-endpoints",
     "tags",
-    "user-type"
+    "user-type",
+    "filtering"
   };
 
   static final Gson gson = new GsonBuilder().serializeNulls().create();
 
@@ -53,6 +53,7 @@ private ComponentsImpl() {}
 
   static final class InMemoryDataStoreFactory implements ComponentConfigurer<DataStore>, DiagnosticDescription {
     static final InMemoryDataStoreFactory INSTANCE = new InMemoryDataStoreFactory();
+
     @Override
     public DataStore build(ClientContext context) {
       return new InMemoryDataStore();
@@ -154,18 +155,17 @@ public DataSource build(ClientContext context) {
           context.getServiceEndpoints().getStreamingBaseUri(),
           StandardEndpoints.DEFAULT_STREAMING_BASE_URI,
           "streaming",
-          baseLogger
-          );
+          baseLogger);
 
       return new StreamProcessor(
           toHttpProperties(context.getHttp()),
           context.getDataSourceUpdateSink(),
           context.getThreadPriority(),
           ClientContextImpl.get(context).diagnosticStore,
           streamUri,
+          payloadFilter,
           initialReconnectDelay,
-          logger
-          );
+          logger);
     }
 
     @Override
@@ -202,17 +202,20 @@ public DataSource build(ClientContext context) {
           context.getServiceEndpoints().getPollingBaseUri(),
           StandardEndpoints.DEFAULT_POLLING_BASE_URI,
           "polling",
-          baseLogger
-          );
+          baseLogger);
+
+      DefaultFeatureRequestor requestor = new DefaultFeatureRequestor(
+          toHttpProperties(context.getHttp()),
+          pollUri,
+          payloadFilter,
+          logger);
 
-      DefaultFeatureRequestor requestor = new DefaultFeatureRequestor(toHttpProperties(context.getHttp()), pollUri, logger);
       return new PollingProcessor(
           requestor,
           context.getDataSourceUpdateSink(),
           ClientContextImpl.get(context).sharedExecutor,
           pollInterval,
-          logger
-          );
+          logger);
     }
 
     @Override
@@ -241,17 +244,15 @@ public EventProcessor build(ClientContext context) {
             null, // use default request path for server-side events
             null, // use default request path for client-side events
             0, // 0 means default retry delay
-            context.getBaseLogger().subLogger(Loggers.EVENTS_LOGGER_NAME)
-            );
+            context.getBaseLogger().subLogger(Loggers.EVENTS_LOGGER_NAME));
       } else {
         eventSender = new EventSenderWrapper(eventSenderConfigurer.build(context));
       }
       URI eventsUri = StandardEndpoints.selectBaseUri(
           context.getServiceEndpoints().getEventsBaseUri(),
           StandardEndpoints.DEFAULT_EVENTS_BASE_URI,
           "events",
-          context.getBaseLogger()
-          );
+          context.getBaseLogger());
       EventsConfiguration eventsConfig = new EventsConfiguration(
           allAttributesPrivate,
           capacity,
@@ -264,8 +265,7 @@ public EventProcessor build(ClientContext context) {
           flushInterval.toMillis(),
           false,
           false,
-          privateAttributes
-          );
+          privateAttributes);
       return new DefaultEventProcessorWrapper(context, eventsConfig);
     }
 
@@ -350,8 +350,7 @@ public HttpConfiguration build(ClientContext clientContext) {
           socketFactory,
           socketTimeout,
           sslSocketFactory,
-          trustManager
-      );
+          trustManager);
     }
   }
 
@@ -378,7 +377,7 @@ public PersistentDataStoreBuilderImpl(ComponentConfigurer<PersistentDataStore> s
     @Override
     public LDValue describeConfiguration(ClientContext clientContext) {
       if (persistentDataStoreConfigurer instanceof DiagnosticDescription) {
-        return ((DiagnosticDescription)persistentDataStoreConfigurer).describeConfiguration(clientContext);
+        return ((DiagnosticDescription) persistentDataStoreConfigurer).describeConfiguration(clientContext);
       }
       return LDValue.of("custom");
     }
@@ -393,8 +392,7 @@ public DataStore build(ClientContext context) {
           recordCacheStats,
           context.getDataStoreUpdateSink(),
           ClientContextImpl.get(context).sharedExecutor,
-          context.getBaseLogger().subLogger(Loggers.DATA_STORE_LOGGER_NAME)
-          );
+          context.getBaseLogger().subLogger(Loggers.DATA_STORE_LOGGER_NAME));
     }
   }
 
@@ -434,8 +432,7 @@ public ServiceEndpoints createServiceEndpoints() {
         return new ServiceEndpoints(
           StandardEndpoints.DEFAULT_STREAMING_BASE_URI,
           StandardEndpoints.DEFAULT_POLLING_BASE_URI,
-          StandardEndpoints.DEFAULT_EVENTS_BASE_URI
-        );
+          StandardEndpoints.DEFAULT_EVENTS_BASE_URI);
       }
       return new ServiceEndpoints(streamingBaseUri, pollingBaseUri, eventsBaseUri);
     }
@@ -455,7 +452,6 @@ static HttpProperties toHttpProperties(HttpConfiguration httpConfig) {
         httpConfig.getSocketFactory(),
         httpConfig.getSocketTimeout().toMillis(),
         httpConfig.getSslSocketFactory(),
-        httpConfig.getTrustManager()
-        );
+        httpConfig.getTrustManager());
   }
 }
@@ -1,22 +1,25 @@
 package com.launchdarkly.sdk.server;
 
+import static com.launchdarkly.sdk.server.DataModelSerialization.parseFullDataSet;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import javax.annotation.Nullable;
+
 import com.google.common.annotations.VisibleForTesting;
 import com.google.gson.stream.JsonReader;
 import com.launchdarkly.logging.LDLogger;
+import com.launchdarkly.sdk.internal.http.HttpConsts;
 import com.launchdarkly.sdk.internal.http.HttpErrors.HttpErrorException;
 import com.launchdarkly.sdk.internal.http.HttpHelpers;
 import com.launchdarkly.sdk.internal.http.HttpProperties;
 import com.launchdarkly.sdk.server.subsystems.DataStoreTypes.FullDataSet;
 import com.launchdarkly.sdk.server.subsystems.DataStoreTypes.ItemDescriptor;
 import com.launchdarkly.sdk.server.subsystems.SerializationException;
 
-import java.io.IOException;
-import java.net.URI;
-import java.nio.file.Files;
-import java.nio.file.Path;
-
-import static com.launchdarkly.sdk.server.DataModelSerialization.parseFullDataSet;
-
 import okhttp3.Cache;
 import okhttp3.Headers;
 import okhttp3.OkHttpClient;
@@ -28,19 +31,36 @@
  */
 final class DefaultFeatureRequestor implements FeatureRequestor {
   private static final long MAX_HTTP_CACHE_SIZE_BYTES = 10 * 1024 * 1024; // 10 MB
-  
-  @VisibleForTesting final URI baseUri;
+
   private final OkHttpClient httpClient;
-  private final URI pollingUri;
+  @VisibleForTesting
+  final URI pollingUri;
   private final Headers headers;
   private final Path cacheDir;
   private final LDLogger logger;
 
-  DefaultFeatureRequestor(HttpProperties httpProperties, URI baseUri, LDLogger logger) {
-    this.baseUri = baseUri;
-    this.pollingUri = HttpHelpers.concatenateUriPath(baseUri, StandardEndpoints.POLLING_REQUEST_PATH);
+  /**
+   * Creates a {@link DefaultFeatureRequestor}
+   * 
+   * @param httpProperties that will be used
+   * @param baseUri        that will be used
+   * @param payloadFilter  identifier that will be used to filter objects in the
+   *                       payload, provide null for no filtering
+   * @param logger         to log with
+   */
+  DefaultFeatureRequestor(HttpProperties httpProperties, URI baseUri, @Nullable String payloadFilter, LDLogger logger) {
     this.logger = logger;
-    
+
+    URI tempUri = HttpHelpers.concatenateUriPath(baseUri, StandardEndpoints.POLLING_REQUEST_PATH);
+    if (payloadFilter != null) {
+      if (!payloadFilter.isEmpty()) {
+        tempUri = HttpHelpers.addQueryParam(tempUri, HttpConsts.QUERY_PARAM_FILTER, payloadFilter);
+      } else {
+        logger.info("Payload filter \"{}\" is not valid, not applying filter.", payloadFilter);
+      }
+    }
+    this.pollingUri = tempUri;
+
     OkHttpClient.Builder httpBuilder = httpProperties.toHttpClientBuilder();
     this.headers = httpProperties.toHeadersBuilder().build();
 
@@ -59,7 +79,7 @@ public void close() {
     HttpProperties.shutdownHttpClient(httpClient);
     Util.deleteDirectory(cacheDir);
   }
-  
+
   public FullDataSet<ItemDescriptor> getAllData(boolean returnDataEvenIfCached)
       throws IOException, HttpErrorException, SerializationException {
     Request request = new Request.Builder()
 
@@ -19,6 +19,7 @@
 import com.launchdarkly.logging.LDLogger;
 import com.launchdarkly.logging.LogValues;
 import com.launchdarkly.sdk.internal.events.DiagnosticStore;
+import com.launchdarkly.sdk.internal.http.HttpConsts;
 import com.launchdarkly.sdk.internal.http.HttpHelpers;
 import com.launchdarkly.sdk.internal.http.HttpProperties;
 import com.launchdarkly.sdk.server.StreamProcessorEvents.DeleteData;
@@ -84,8 +85,10 @@ final class StreamProcessor implements DataSource {
   private final DataSourceUpdateSink dataSourceUpdates;
   private final HttpProperties httpProperties;
   private final Headers headers;
-  @VisibleForTesting final URI streamUri;
-  @VisibleForTesting final Duration initialReconnectDelay;
+  @VisibleForTesting
+  final URI streamUri;
+  @VisibleForTesting
+  final Duration initialReconnectDelay;
   private final DiagnosticStore diagnosticAccumulator;
   private final int threadPriority;
   private final DataStoreStatusProvider.StatusListener statusListener;
@@ -102,17 +105,27 @@ final class StreamProcessor implements DataSource {
       int threadPriority,
       DiagnosticStore diagnosticAccumulator,
       URI streamUri,
+      String payloadFilter,
       Duration initialReconnectDelay,
-      LDLogger logger
-      ) {
+      LDLogger logger) {
     this.dataSourceUpdates = dataSourceUpdates;
     this.httpProperties = httpProperties;
     this.diagnosticAccumulator = diagnosticAccumulator;
     this.threadPriority = threadPriority;
-    this.streamUri = streamUri;
     this.initialReconnectDelay = initialReconnectDelay;
     this.logger = logger;
 
+    URI tempUri = HttpHelpers.concatenateUriPath(streamUri, StandardEndpoints.STREAMING_REQUEST_PATH);
+    if (payloadFilter != null) {
+      if (!payloadFilter.isEmpty()) {
+        tempUri = HttpHelpers.addQueryParam(tempUri, HttpConsts.QUERY_PARAM_FILTER, payloadFilter);
+      } else {
+        logger.info("Payload filter \"{}\" is not valid, not applying filter.", payloadFilter);
+      }
+    }
+    this.streamUri = tempUri;
+
+
     this.headers = httpProperties.toHeadersBuilder()
         .add("Accept", "text/event-stream")
         .build();
@@ -144,8 +157,6 @@ private void onStoreStatusChanged(DataStoreStatusProvider.Status newStatus) {
   public Future<Void> start() {
     final CompletableFuture<Void> initFuture = new CompletableFuture<>();
 
-    URI endpointUri = HttpHelpers.concatenateUriPath(streamUri, StandardEndpoints.STREAMING_REQUEST_PATH);
-
     // Notes about the configuration of the EventSource below:
     //
     // 1. Setting streamEventData(true) is an optimization to let us read the event's data field directly
@@ -158,7 +169,7 @@ public Future<Void> start() {
     // LD client gets. A read timeout on the stream will result in the connection being cycled, so we set
     // this to be slightly more than the expected interval between heartbeat signals.
 
-    HttpConnectStrategy eventSourceHttpConfig = ConnectStrategy.http(endpointUri)
+    HttpConnectStrategy eventSourceHttpConfig = ConnectStrategy.http(this.streamUri)
         .headers(headers)
         .clientBuilderActions(clientBuilder -> {
           httpProperties.applyToHttpClientBuilder(clientBuilder);
 
@@ -34,6 +34,8 @@ public abstract class PollingDataSourceBuilder implements ComponentConfigurer<Da
   public static final Duration DEFAULT_POLL_INTERVAL = Duration.ofSeconds(30);
 
   protected Duration pollInterval = DEFAULT_POLL_INTERVAL;
+
+  protected String payloadFilter;
 
   /**
    * Sets the interval at which the SDK will poll for feature flag updates.
@@ -52,4 +54,16 @@ public PollingDataSourceBuilder pollInterval(Duration pollInterval) {
     }
     return this;
   }
+
+  /**
+   * Sets the Payload Filter that will be used to filter the objects (flags, segments, etc.)
+   * from this data source.
+   * 
+   * @param payloadFilter the filter to be used
+   * @return the builder
+   */
+  public PollingDataSourceBuilder payloadFilter(String payloadFilter) {
+    this.payloadFilter = payloadFilter;
+    return this;
+  }
 }
@@ -1,12 +1,12 @@
 package com.launchdarkly.sdk.server.integrations;
 
+import java.time.Duration;
+
 import com.launchdarkly.sdk.server.Components;
 import com.launchdarkly.sdk.server.LDConfig.Builder;
 import com.launchdarkly.sdk.server.subsystems.ComponentConfigurer;
 import com.launchdarkly.sdk.server.subsystems.DataSource;
 
-import java.time.Duration;
-
 /**
  * Contains methods for configuring the streaming data source.
  * <p>
@@ -31,6 +31,8 @@ public abstract class StreamingDataSourceBuilder implements ComponentConfigurer<
 
   protected Duration initialReconnectDelay = DEFAULT_INITIAL_RECONNECT_DELAY;
 
+  protected String payloadFilter;
+
   /**
    * Sets the initial reconnect delay for the streaming connection.
    * <p>
@@ -48,4 +50,16 @@ public StreamingDataSourceBuilder initialReconnectDelay(Duration initialReconnec
     this.initialReconnectDelay = initialReconnectDelay == null ? DEFAULT_INITIAL_RECONNECT_DELAY : initialReconnectDelay;
     return this;
   }
+
+  /**
+   * Sets the Payload Filter that will be used to filter the objects (flags, segments, etc.)
+   * from this data source.
+   * 
+   * @param payloadFilter the filter to be used
+   * @return the builder
+   */
+  public StreamingDataSourceBuilder payloadFilter(String payloadFilter) {
+    this.payloadFilter = payloadFilter;
+    return this;
+  }
 }
Original file line number	Diff line number	Diff line change
`@@ -35,6 +35,7 @@ public static class SdkConfigParams {`
`35`	`35`	`public static class SdkConfigStreamParams {`
`36`	`36`	`URI baseUri;`
`37`	`37`	`long initialRetryDelayMs;`
	`38`	`+ String filter;`
`38`	`39`	`}`
`39`	`40`
`40`	`41`	`public static class SdkConfigEventParams {`
Original file line number	Diff line number	Diff line change
`@@ -300,6 +300,7 @@ private LDConfig buildSdkConfig(SdkConfigParams params, String tag) {`
`300`	`300`	`if (params.streaming.initialRetryDelayMs > 0) {`
`301`	`301`	`dataSource.initialReconnectDelay(Duration.ofMillis(params.streaming.initialRetryDelayMs));`
`302`	`302`	`}`
	`303`	`+ dataSource.payloadFilter(params.streaming.filter);`
`303`	`304`	`builder.dataSource(dataSource);`
`304`	`305`	`}`
`305`	`306`