chore: refactor flagd event handling (#398)

Signed-off-by: Kavindu Dodanduwa <[email protected]>
Signed-off-by: Kavindu Dodanduwa <[email protected]>
Co-authored-by: Giovanni Liva <[email protected]>

Author: Kavindu-Dodan

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

@@ -1,16 +1,7 @@
 package dev.openfeature.contrib.providers.flagd;
 
-import dev.openfeature.contrib.providers.flagd.cache.Cache;
 import dev.openfeature.contrib.providers.flagd.cache.CacheFactory;
-import dev.openfeature.contrib.providers.flagd.grpc.FlagResolution;
-import dev.openfeature.contrib.providers.flagd.grpc.GrpcConnector;
-import dev.openfeature.contrib.providers.flagd.strategy.ResolveFactory;
-import dev.openfeature.contrib.providers.flagd.strategy.ResolveStrategy;
-import dev.openfeature.flagd.grpc.Schema.ResolveBooleanRequest;
-import dev.openfeature.flagd.grpc.Schema.ResolveFloatRequest;
-import dev.openfeature.flagd.grpc.Schema.ResolveIntRequest;
-import dev.openfeature.flagd.grpc.Schema.ResolveObjectRequest;
-import dev.openfeature.flagd.grpc.Schema.ResolveStringRequest;
+import dev.openfeature.contrib.providers.flagd.grpc.GrpcResolution;
 import dev.openfeature.sdk.EvaluationContext;
 import dev.openfeature.sdk.EventProvider;
 import dev.openfeature.sdk.FeatureProvider;
@@ -34,12 +25,7 @@ public class FlagdProvider extends EventProvider implements FeatureProvider {
     private static final String FLAGD_PROVIDER = "flagD Provider";
 
     private final ReadWriteLock lock = new ReentrantReadWriteLock();
-
-    private final Cache cache;
-    private final ResolveStrategy strategy;
-    private final GrpcConnector grpc;
-    private final FlagResolution flagResolver;
-
+    private final Resolver flagResolver;
 
     private ProviderState state = ProviderState.NOT_READY;
 
@@ -56,28 +42,18 @@ public FlagdProvider() {
      * @param options {@link FlagdOptions} with
      */
     public FlagdProvider(final FlagdOptions options) {
-        this.strategy = ResolveFactory.getStrategy(options);
-        this.cache = CacheFactory.getCache(options);
-        this.grpc = new GrpcConnector(options, this.cache, this::setState);
-        this.flagResolver = new FlagResolution(this.cache, this.strategy, this::getState);
-    }
-
-    FlagdProvider(ResolveStrategy strategy, Cache cache, GrpcConnector grpc, FlagResolution resolution) {
-        this.strategy = strategy;
-        this.cache = cache;
-        this.grpc = grpc;
-        this.flagResolver = resolution;
+        this.flagResolver = new GrpcResolution(options, CacheFactory.getCache(options), this::getState, this::setState);
     }
 
     @Override
-    public void initialize(EvaluationContext evaluationContext) throws RuntimeException {
-        this.grpc.initialize();
+    public void initialize(EvaluationContext evaluationContext) throws Exception {
+        this.flagResolver.init();
     }
 
     @Override
     public void shutdown() {
         try {
-            this.grpc.shutdown();
+            this.flagResolver.shutdown();
         } catch (Exception e) {
             log.error("Error during shutdown {}", FLAGD_PROVIDER, e);
         }
@@ -94,6 +70,37 @@ public ProviderState getState() {
         }
     }
 
+
+    @Override
+    public Metadata getMetadata() {
+        return () -> FLAGD_PROVIDER;
+    }
+
+    @Override
+    public ProviderEvaluation<Boolean> getBooleanEvaluation(String key, Boolean defaultValue, EvaluationContext ctx) {
+        return this.flagResolver.booleanEvaluation(key, defaultValue, ctx);
+    }
+
+    @Override
+    public ProviderEvaluation<String> getStringEvaluation(String key, String defaultValue, EvaluationContext ctx) {
+        return this.flagResolver.stringEvaluation(key, defaultValue, ctx);
+    }
+
+    @Override
+    public ProviderEvaluation<Double> getDoubleEvaluation(String key, Double defaultValue, EvaluationContext ctx) {
+        return this.flagResolver.doubleEvaluation(key, defaultValue, ctx);
+    }
+
+    @Override
+    public ProviderEvaluation<Integer> getIntegerEvaluation(String key, Integer defaultValue, EvaluationContext ctx) {
+        return this.flagResolver.integerEvaluation(key, defaultValue, ctx);
+    }
+
+    @Override
+    public ProviderEvaluation<Value> getObjectEvaluation(String key, Value defaultValue, EvaluationContext ctx) {
+        return this.flagResolver.objectEvaluation(key, defaultValue, ctx);
+    }
+
     private void setState(ProviderState newState) {
         ProviderState oldState;
         Lock l = this.lock.writeLock();
@@ -108,10 +115,6 @@ private void setState(ProviderState newState) {
     }
 
     private void handleStateTransition(ProviderState oldState, ProviderState newState) {
-        // we are connected, reset the gRPC retry logic
-        if (ProviderState.READY.equals(newState)) {
-            this.grpc.resetRetryConnection();
-        }
         // we got initialized
         if (ProviderState.NOT_READY.equals(oldState) && ProviderState.READY.equals(newState)) {
             // nothing to do, the SDK emits the events
@@ -140,61 +143,4 @@ private void handleStateTransition(ProviderState oldState, ProviderState newStat
             this.emitProviderConfigurationChanged(details);
         }
     }
-
-
-    @Override
-    public Metadata getMetadata() {
-        return () -> FLAGD_PROVIDER;
-    }
-
-    @Override
-    public ProviderEvaluation<Boolean> getBooleanEvaluation(String key, Boolean defaultValue,
-                                                            EvaluationContext ctx) {
-
-        ResolveBooleanRequest request = ResolveBooleanRequest.newBuilder().buildPartial();
-
-        return this.flagResolver.resolve(key, ctx, request,
-                this.grpc.getResolver()::resolveBoolean, null);
-    }
-
-    @Override
-    public ProviderEvaluation<String> getStringEvaluation(String key, String defaultValue,
-                                                          EvaluationContext ctx) {
-        ResolveStringRequest request = ResolveStringRequest.newBuilder().buildPartial();
-
-        return this.flagResolver.resolve(key, ctx, request,
-                this.grpc.getResolver()::resolveString, null);
-    }
-
-    @Override
-    public ProviderEvaluation<Double> getDoubleEvaluation(String key, Double defaultValue,
-                                                          EvaluationContext ctx) {
-        ResolveFloatRequest request = ResolveFloatRequest.newBuilder().buildPartial();
-
-        return this.flagResolver.resolve(key, ctx, request,
-                this.grpc.getResolver()::resolveFloat, null);
-    }
-
-    @Override
-    public ProviderEvaluation<Integer> getIntegerEvaluation(String key, Integer defaultValue,
-                                                            EvaluationContext ctx) {
-
-        ResolveIntRequest request = ResolveIntRequest.newBuilder().buildPartial();
-
-        return this.flagResolver.resolve(key, ctx, request,
-                this.grpc.getResolver()::resolveInt,
-                (Object value) -> ((Long) value).intValue());
-    }
-
-    @Override
-    public ProviderEvaluation<Value> getObjectEvaluation(String key, Value defaultValue,
-                                                         EvaluationContext ctx) {
-
-        ResolveObjectRequest request = ResolveObjectRequest.newBuilder().buildPartial();
-
-        return this.flagResolver.resolve(key, ctx, request,
-                this.grpc.getResolver()::resolveObject,
-                (Object value) -> this.flagResolver.convertObjectResponse((com.google.protobuf.Struct) value));
-    }
-
 }

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

@@ -0,0 +1,24 @@
+package dev.openfeature.contrib.providers.flagd;
+
+import dev.openfeature.sdk.EvaluationContext;
+import dev.openfeature.sdk.ProviderEvaluation;
+import dev.openfeature.sdk.Value;
+
+/**
+ * A generic contract flag resolving contract for flagd.
+ */
+public interface Resolver {
+    void init() throws Exception;
+
+    void shutdown() throws Exception;
+
+    ProviderEvaluation<Boolean> booleanEvaluation(String key, Boolean defaultValue, EvaluationContext ctx);
+
+    ProviderEvaluation<String> stringEvaluation(String key, String defaultValue, EvaluationContext ctx);
+
+    ProviderEvaluation<Double> doubleEvaluation(String key, Double defaultValue, EvaluationContext ctx);
+
+    ProviderEvaluation<Integer> integerEvaluation(String key, Integer defaultValue, EvaluationContext ctx);
+
+    ProviderEvaluation<Value> objectEvaluation(String key, Value defaultValue, EvaluationContext ctx);
+}

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/grpc/EventStreamCallback.java

@@ -16,9 +16,9 @@
  */
 @Slf4j
 @SuppressFBWarnings(justification = "cache needs to be read and write by multiple objects")
-public class EventStreamObserver implements StreamObserver<EventStreamResponse> {
+class EventStreamObserver implements StreamObserver<EventStreamResponse> {
     private final Consumer<ProviderState> stateConsumer;
-    private final Runnable reconnectEventStream;
+    private final Object sync;
     private final Cache cache;
 
     private static final String CONFIGURATION_CHANGE = "configuration_change";
@@ -27,14 +27,15 @@ public class EventStreamObserver implements StreamObserver<EventStreamResponse>
 
     /**
      * Create a gRPC stream that get notified about flag changes.
-     * @param cache cache to update
-     * @param stateConsumer lambda to call for setting the state
-     * @param reconnectEventStream callback for trying to recreate the stream
+     *
+     * @param sync                 synchronization object from caller
+     * @param cache                cache to update
+     * @param stateConsumer        lambda to call for setting the state
      */
-    public EventStreamObserver(Cache cache, Consumer<ProviderState> stateConsumer, Runnable reconnectEventStream) {
+    EventStreamObserver(Object sync, Cache cache, Consumer<ProviderState> stateConsumer) {
+        this.sync = sync;
         this.cache = cache;
         this.stateConsumer = stateConsumer;
-        this.reconnectEventStream = reconnectEventStream;
     }
 
     @Override
@@ -58,7 +59,9 @@ public void onError(Throwable t) {
             this.cache.clear();
         }
         this.stateConsumer.accept(ProviderState.ERROR);
-        this.reconnectEventStream.run();
+
+        // handle last call of this stream
+        handleEndOfStream();
     }
 
     @Override
@@ -67,6 +70,9 @@ public void onCompleted() {
             this.cache.clear();
         }
         this.stateConsumer.accept(ProviderState.ERROR);
+
+        // handle last call of this stream
+        handleEndOfStream();
     }
 
     private void handleConfigurationChangeEvent(EventStreamResponse value) {
@@ -93,4 +99,10 @@ private void handleProviderReadyEvent() {
             this.cache.clear();
         }
     }
+
+    private void handleEndOfStream() {
+        synchronized (this.sync) {
+            this.sync.notify();
+        }
+    }
 }

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/grpc/EventStreamObserver.java

@@ -18,13 +18,14 @@
  * FlagdGrpcInterceptor is an interceptor for grpc communication from java-sdk to flagd.
  * <a href="https://github.com/open-telemetry/opentelemetry-java-docs">credits</a>
  */
-public final class FlagdGrpcInterceptor implements ClientInterceptor {
+final class FlagdGrpcInterceptor implements ClientInterceptor {
     private static final TextMapSetter<Metadata> SETTER = new Setter();
 
     private final OpenTelemetry openTelemetry;
 
     /**
      * Intercept gRPC communication for propagating OpenTelemetry context over the wire.
+     *
      * @param openTelemetry the OpenTelemetry implementation.
      */
     public FlagdGrpcInterceptor(@Nonnull final OpenTelemetry openTelemetry) {