refactor: remove channel monitor (#1670)

Signed-off-by: Todd Baert <[email protected]>

Author: toddbaert

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/common/ChannelConnector.java

@@ -1,19 +1,13 @@
 package dev.openfeature.contrib.providers.flagd.resolver.common;
 
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
-import dev.openfeature.sdk.ImmutableStructure;
-import dev.openfeature.sdk.ProviderEvent;
-import io.grpc.ConnectivityState;
 import io.grpc.ManagedChannel;
-import java.util.Collections;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
 /**
- * A generic GRPC connector that manages connection states, reconnection logic, and event streaming for
- * GRPC services.
+ * A GRPC connector that maintains a managed channel for communication with a flagd server and handles shutdown.
  */
 @Slf4j
 public class ChannelConnector {
@@ -29,33 +23,15 @@ public class ChannelConnector {
      */
     private final long deadline;
 
-    /**
-     * A consumer that handles connection events such as connection loss or reconnection.
-     */
-    private final Consumer<FlagdProviderEvent> onConnectionEvent;
-
     /**
      * Constructs a new {@code ChannelConnector} instance with the specified options and parameters.
      *
      * @param options             the configuration options for the GRPC connection
-     * @param onConnectionEvent   a consumer to handle connection events
      * @param channel             the managed channel for the GRPC connection
      */
-    public ChannelConnector(
-            final FlagdOptions options, final Consumer<FlagdProviderEvent> onConnectionEvent, ManagedChannel channel) {
+    public ChannelConnector(final FlagdOptions options, ManagedChannel channel) {
         this.channel = channel;
         this.deadline = options.getDeadline();
-        this.onConnectionEvent = onConnectionEvent;
-    }
-
-    /**
-     * Initializes the GRPC connection by waiting for the channel to be ready and monitoring its state.
-     *
-     * @throws Exception if the channel does not reach the desired state within the deadline
-     */
-    public void initialize() throws Exception {
-        log.info("Initializing GRPC connection.");
-        monitorChannelState(ConnectivityState.READY);
     }
 
     /**
@@ -71,27 +47,4 @@ public void shutdown() throws InterruptedException {
             channel.awaitTermination(deadline, TimeUnit.MILLISECONDS);
         }
     }
-
-    /**
-     * Monitors the state of a gRPC channel and triggers the specified callbacks based on state changes.
-     *
-     * @param expectedState     the initial state to monitor.
-     */
-    private void monitorChannelState(ConnectivityState expectedState) {
-        channel.notifyWhenStateChanged(expectedState, this::onStateChange);
-    }
-
-    private void onStateChange() {
-        ConnectivityState currentState = channel.getState(true);
-        log.debug("Channel state changed to: {}", currentState);
-        if (currentState == ConnectivityState.TRANSIENT_FAILURE || currentState == ConnectivityState.SHUTDOWN) {
-            this.onConnectionEvent.accept(new FlagdProviderEvent(
-                    ProviderEvent.PROVIDER_ERROR, Collections.emptyList(), new ImmutableStructure()));
-        }
-        if (currentState != ConnectivityState.SHUTDOWN) {
-            log.debug("continuing to monitor the grpc channel");
-            // Re-register the state monitor to watch for the next state transition.
-            monitorChannelState(currentState);
-        }
-    }
 }

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

@@ -52,7 +52,7 @@ public class InProcessResolver implements Resolver {
      *                          connection/stream
      */
     public InProcessResolver(FlagdOptions options, Consumer<FlagdProviderEvent> onConnectionEvent) {
-        this.flagStore = new FlagStore(getConnector(options, onConnectionEvent));
+        this.flagStore = new FlagStore(getConnector(options));
         this.onConnectionEvent = onConnectionEvent;
         this.operator = new Operator();
         this.scope = options.getSelector();
@@ -147,14 +147,14 @@ public ProviderEvaluation<Value> objectEvaluation(String key, Value defaultValue
                 .build();
     }
 
-    static QueueSource getConnector(final FlagdOptions options, Consumer<FlagdProviderEvent> onConnectionEvent) {
+    static QueueSource getConnector(final FlagdOptions options) {
         if (options.getCustomConnector() != null) {
             return options.getCustomConnector();
         }
         return options.getOfflineFlagSourcePath() != null
                         && !options.getOfflineFlagSourcePath().isEmpty()
                 ? new FileQueueSource(options.getOfflineFlagSourcePath(), options.getOfflinePollIntervalMs())
-                : new SyncStreamQueueSource(options, onConnectionEvent);
+                : new SyncStreamQueueSource(options);
     }
 
     private <T> ProviderEvaluation<T> resolve(Class<T> type, String key, EvaluationContext ctx) {

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSource.java

@@ -4,7 +4,6 @@
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
 import dev.openfeature.contrib.providers.flagd.resolver.common.ChannelBuilder;
 import dev.openfeature.contrib.providers.flagd.resolver.common.ChannelConnector;
-import dev.openfeature.contrib.providers.flagd.resolver.common.FlagdProviderEvent;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueuePayload;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueuePayloadType;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueueSource;
@@ -24,7 +23,6 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -53,14 +51,14 @@ public class SyncStreamQueueSource implements QueueSource {
     /**
      * Creates a new SyncStreamQueueSource responsible for observing the event stream.
      */
-    public SyncStreamQueueSource(final FlagdOptions options, Consumer<FlagdProviderEvent> onConnectionEvent) {
+    public SyncStreamQueueSource(final FlagdOptions options) {
         streamDeadline = options.getStreamDeadlineMs();
         deadline = options.getDeadline();
         selector = options.getSelector();
         providerId = options.getProviderId();
         maxBackoffMs = options.getRetryBackoffMaxMs();
         syncMetadataDisabled = options.isSyncMetadataDisabled();
-        channelConnector = new ChannelConnector(options, onConnectionEvent, ChannelBuilder.nettyChannel(options));
+        channelConnector = new ChannelConnector(options, ChannelBuilder.nettyChannel(options));
         flagSyncStub =
                 FlagSyncServiceGrpc.newStub(channelConnector.getChannel()).withWaitForReady();
         metadataStub = FlagSyncServiceGrpc.newBlockingStub(channelConnector.getChannel())
@@ -86,7 +84,6 @@ protected SyncStreamQueueSource(
 
     /** Initialize sync stream connector. */
     public void init() throws Exception {
-        channelConnector.initialize();
         Thread listener = new Thread(this::observeSyncStream);
         listener.setDaemon(true);
         listener.start();

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/rpc/RpcResolver.java

@@ -84,7 +84,7 @@ public RpcResolver(
         this.strategy = ResolveFactory.getStrategy(options);
         this.options = options;
         incomingQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
-        this.connector = new ChannelConnector(options, onProviderEvent, ChannelBuilder.nettyChannel(options));
+        this.connector = new ChannelConnector(options, ChannelBuilder.nettyChannel(options));
         this.onProviderEvent = onProviderEvent;
         this.stub = ServiceGrpc.newStub(this.connector.getChannel()).withWaitForReady();
         this.blockingStub =
@@ -113,8 +113,6 @@ protected RpcResolver(
      * Initialize RpcResolver resolver.
      */
     public void init() throws Exception {
-        this.connector.initialize();
-
         Thread listener = new Thread(() -> {
             try {
                 observeEventStream();

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/common/ChannelConnectorTest.java

@@ -70,9 +70,9 @@ public void connectorSetup() {
                 .build();
 
         // then
-        assertInstanceOf(SyncStreamQueueSource.class, InProcessResolver.getConnector(forGrpcOptions, e -> {}));
-        assertInstanceOf(FileQueueSource.class, InProcessResolver.getConnector(forOfflineOptions, e -> {}));
-        assertInstanceOf(MockConnector.class, InProcessResolver.getConnector(forCustomConnectorOptions, e -> {}));
+        assertInstanceOf(SyncStreamQueueSource.class, InProcessResolver.getConnector(forGrpcOptions));
+        assertInstanceOf(FileQueueSource.class, InProcessResolver.getConnector(forOfflineOptions));
+        assertInstanceOf(MockConnector.class, InProcessResolver.getConnector(forCustomConnectorOptions));
     }
 
     @Test

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

@@ -7,7 +7,6 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -48,7 +47,6 @@ public void setup() throws Exception {
         when(blockingStub.getMetadata(any())).thenReturn(GetMetadataResponse.getDefaultInstance());
 
         mockConnector = mock(ChannelConnector.class);
-        doNothing().when(mockConnector).initialize(); // Mock the initialize method
 
         stub = mock(FlagSyncServiceStub.class);
         when(stub.withDeadlineAfter(anyLong(), any())).thenReturn(stub);

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSourceTest.java

@@ -44,7 +44,6 @@ public void init() throws Exception {
         blockingStub = mock(ServiceBlockingStub.class);
 
         mockConnector = mock(ChannelConnector.class);
-        doNothing().when(mockConnector).initialize(); // Mock the initialize method
 
         stub = mock(ServiceStub.class);
         when(stub.withDeadlineAfter(anyLong(), any())).thenReturn(stub);