feat: [flagd-in-process] Support Injection of a custom connector (#900)

Signed-off-by: pratik <[email protected]>

Author: prathyk

providers/flagd/README.md

@@ -64,9 +64,28 @@ Provider will attempt to detect file changes using polling.
 Polling happens at 5 second intervals and this is currently unconfigurable.
 This mode is useful for local development, tests and offline applications.
 
+#### Custom Connector 
+
+You can include a custom connector as a configuration option to customize how the in-process resolver fetches flags. 
+The custom connector must implement the [Connector interface](https://github.com/open-feature/java-sdk-contrib/blob/main/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/Connector.java).
+
+```java
+Connector myCustomConnector = new MyCustomConnector();
+FlagdOptions options =
+        FlagdOptions.builder()
+                .resolverType(Config.Resolver.IN_PROCESS)
+                .customConnector(myCustomConnector)
+                .build();
+
+FlagdProvider flagdProvider = new FlagdProvider(options);
+```
+
 > [!IMPORTANT]
-> Note that you can only use a single flag source (either gRPC or offline file) for the in-process resolver.
-> If both sources are configured, offline mode will be selected.
+> Note that the in-process resolver can only use a single flag source.
+> If multiple sources are configured then only one would be selected based on the following order of preference:
+>   1. Custom Connector
+>   2. Offline file
+>   3. gRPC
 
 ### Configuration options
 

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/FlagdOptions.java

@@ -1,5 +1,6 @@
 package dev.openfeature.contrib.providers.flagd;
 
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.Connector;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
 import lombok.Builder;
@@ -99,6 +100,11 @@ public class FlagdOptions {
     @Builder.Default
     private String offlineFlagSourcePath = fallBackToEnvOrDefault(Config.OFFLINE_SOURCE_PATH, null);
 
+    /**
+     * Inject a Custom Connector for fetching flags.
+     */
+    private Connector customConnector;
+
     /**
      * Inject OpenTelemetry for the library runtime. Providing sdk will initiate
      * distributed tracing for flagd grpc

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/InProcessResolver.java

@@ -152,6 +152,9 @@ public ProviderEvaluation<Value> objectEvaluation(String key, Value defaultValue
     }
 
     static Connector getConnector(final FlagdOptions options) {
+        if (options.getCustomConnector() != null) {
+            return options.getCustomConnector();
+        }
         return options.getOfflineFlagSourcePath() != null && !options.getOfflineFlagSourcePath().isEmpty()
                 ? new FileConnector(options.getOfflineFlagSourcePath())
                 : new GrpcStreamConnector(options);

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/FlagdOptionsTest.java

@@ -1,5 +1,7 @@
 package dev.openfeature.contrib.providers.flagd;
 
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.MockConnector;
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.Connector;
 import io.opentelemetry.api.OpenTelemetry;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
@@ -28,13 +30,15 @@ void TestDefaults() {
         assertEquals(DEFAULT_MAX_EVENT_STREAM_RETRIES, builder.getMaxEventStreamRetries());
         assertNull(builder.getSelector());
         assertNull(builder.getOpenTelemetry());
+        assertNull(builder.getCustomConnector());
         assertNull(builder.getOfflineFlagSourcePath());
         assertEquals(Resolver.RPC, builder.getResolverType());
     }
 
     @Test
     void TestBuilderOptions() {
         OpenTelemetry openTelemetry = Mockito.mock(OpenTelemetry.class);
+        Connector connector = new MockConnector(null);
 
         FlagdOptions flagdOptions = FlagdOptions.builder()
                 .host("https://hosted-flagd")
@@ -47,6 +51,7 @@ void TestBuilderOptions() {
                 .selector("app=weatherApp")
                 .offlineFlagSourcePath("some-path")
                 .openTelemetry(openTelemetry)
+                .customConnector(connector)
                 .resolverType(Resolver.IN_PROCESS)
                 .build();
 
@@ -60,6 +65,7 @@ void TestBuilderOptions() {
         assertEquals("app=weatherApp", flagdOptions.getSelector());
         assertEquals("some-path", flagdOptions.getOfflineFlagSourcePath());
         assertEquals(openTelemetry, flagdOptions.getOpenTelemetry());
+        assertEquals(connector, flagdOptions.getCustomConnector());
         assertEquals(Resolver.IN_PROCESS, flagdOptions.getResolverType());
     }
 

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/InProcessResolverTest.java

@@ -3,6 +3,7 @@
 import dev.openfeature.contrib.providers.flagd.Config;
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
 import dev.openfeature.contrib.providers.flagd.resolver.process.model.FeatureFlag;
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.MockConnector;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.StorageState;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.file.FileConnector;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.grpc.GrpcStreamConnector;
@@ -53,10 +54,13 @@ public void connectorSetup(){
                  FlagdOptions.builder().resolverType(Config.Resolver.IN_PROCESS).host("localhost").port(8080).build();
         FlagdOptions forOfflineOptions =
                 FlagdOptions.builder().resolverType(Config.Resolver.IN_PROCESS).offlineFlagSourcePath("path").build();
+        FlagdOptions forCustomConnectorOptions =
+                FlagdOptions.builder().resolverType(Config.Resolver.IN_PROCESS).customConnector(new MockConnector(null)).build();
 
         // then
         assertInstanceOf(GrpcStreamConnector.class, InProcessResolver.getConnector(forGrpcOptions));
         assertInstanceOf(FileConnector.class, InProcessResolver.getConnector(forOfflineOptions));
+        assertInstanceOf(MockConnector.class, InProcessResolver.getConnector(forCustomConnectorOptions));
     }
 
     @Test

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/MockConnector.java

@@ -12,7 +12,7 @@ public class MockConnector implements Connector {
 
     private BlockingQueue<StreamPayload> mockQueue;
 
-    MockConnector(final BlockingQueue<StreamPayload> mockQueue) {
+    public MockConnector(final BlockingQueue<StreamPayload> mockQueue) {
         this.mockQueue = mockQueue;
     }