fix(flagd): Simplified ChannelConnector

Signed-off-by: Guido Breitenhuber <[email protected]>

Author: guidobrei

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

@@ -16,12 +16,9 @@
 /**
  * A generic GRPC connector that manages connection states, reconnection logic, and event streaming for
  * GRPC services.
- *
- * @param <T> the type of the asynchronous stub for the GRPC service
- * @param <K> the type of the blocking stub for the GRPC service
  */
 @Slf4j
-public class ChannelConnector<T extends AbstractStub<T>, K extends AbstractBlockingStub<K>> {
+public class ChannelConnector {
 
     /**
      * The GRPC managed channel for managing the underlying GRPC connection.
@@ -48,22 +45,11 @@ public class ChannelConnector<T extends AbstractStub<T>, K extends AbstractBlock
      */
     public ChannelConnector(
             final FlagdOptions options, final Consumer<FlagdProviderEvent> onConnectionEvent, ManagedChannel channel) {
-
         this.channel = channel;
         this.deadline = options.getDeadline();
         this.onConnectionEvent = onConnectionEvent;
     }
 
-    /**
-     * Constructs a {@code ChannelConnector} instance for testing purposes.
-     *
-     * @param options             the configuration options for the GRPC connection
-     * @param onConnectionEvent   a consumer to handle connection events
-     */
-    public ChannelConnector(final FlagdOptions options, final Consumer<FlagdProviderEvent> onConnectionEvent) {
-        this(options, onConnectionEvent, ChannelBuilder.nettyChannel(options));
-    }
-
     /**
      * Initializes the GRPC connection by waiting for the channel to be ready and monitoring its state.
      *

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

@@ -2,6 +2,7 @@
 
 import com.google.protobuf.Struct;
 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;
@@ -17,7 +18,6 @@
 import dev.openfeature.sdk.Awaitable;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.grpc.Status;
-import io.grpc.StatusException;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
 import java.util.concurrent.BlockingQueue;
@@ -43,10 +43,10 @@ public class SyncStreamQueueSource implements QueueSource {
     private final String selector;
     private final String providerId;
     private final boolean syncMetadataDisabled;
-    private final ChannelConnector<FlagSyncServiceStub, FlagSyncServiceBlockingStub> channelConnector;
+    private final ChannelConnector channelConnector;
     private final BlockingQueue<QueuePayload> outgoingQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
-    private final FlagSyncServiceStub stub;
-    private final FlagSyncServiceBlockingStub blockingStub;
+    private final FlagSyncServiceStub flagSyncStub;
+    private final FlagSyncServiceBlockingStub metadataStub;
 
     /**
      * Creates a new SyncStreamQueueSource responsible for observing the event stream.
@@ -57,26 +57,25 @@ public SyncStreamQueueSource(final FlagdOptions options, Consumer<FlagdProviderE
         selector = options.getSelector();
         providerId = options.getProviderId();
         syncMetadataDisabled = options.isSyncMetadataDisabled();
-        channelConnector = new ChannelConnector<>(options, onConnectionEvent);
-        this.stub = FlagSyncServiceGrpc.newStub(channelConnector.getChannel()).withWaitForReady();
-        this.blockingStub = FlagSyncServiceGrpc.newBlockingStub(channelConnector.getChannel())
-                .withWaitForReady();
+        channelConnector = new ChannelConnector(options, onConnectionEvent, ChannelBuilder.nettyChannel(options));
+        flagSyncStub = FlagSyncServiceGrpc.newStub(channelConnector.getChannel()).withWaitForReady();
+        metadataStub = FlagSyncServiceGrpc.newBlockingStub(channelConnector.getChannel()).withWaitForReady();
     }
 
     // internal use only
     protected SyncStreamQueueSource(
             final FlagdOptions options,
-            ChannelConnector<FlagSyncServiceStub, FlagSyncServiceBlockingStub> connectorMock,
+            ChannelConnector connectorMock,
             FlagSyncServiceStub stubMock,
             FlagSyncServiceBlockingStub blockingStubMock) {
         streamDeadline = options.getStreamDeadlineMs();
         deadline = options.getDeadline();
         selector = options.getSelector();
         providerId = options.getProviderId();
         channelConnector = connectorMock;
-        stub = stubMock;
+        flagSyncStub = stubMock;
         syncMetadataDisabled = options.isSyncMetadataDisabled();
-        blockingStub = blockingStubMock;
+        metadataStub = blockingStubMock;
     }
 
     /** Initialize sync stream connector. */
@@ -141,7 +140,7 @@ private Struct getMetadata() {
             return null;
         }
 
-        FlagSyncServiceBlockingStub localStub = blockingStub;
+        FlagSyncServiceBlockingStub localStub = metadataStub;
 
         if (deadline > 0) {
             localStub = localStub.withDeadlineAfter(deadline, TimeUnit.MILLISECONDS);
@@ -167,7 +166,7 @@ private Struct getMetadata() {
     }
 
     private void syncFlags(SyncStreamObserver streamObserver) {
-        FlagSyncServiceStub localStub = stub; // don't mutate the stub
+        FlagSyncServiceStub localStub = flagSyncStub; // don't mutate the stub
         if (streamDeadline > 0) {
             localStub = localStub.withDeadlineAfter(streamDeadline, TimeUnit.MILLISECONDS);
         }

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

@@ -9,6 +9,7 @@
 import dev.openfeature.contrib.providers.flagd.Config;
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
 import dev.openfeature.contrib.providers.flagd.resolver.Resolver;
+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.common.QueueingStreamObserver;
@@ -59,7 +60,7 @@
 public final class RpcResolver implements Resolver {
     private static final int QUEUE_SIZE = 5;
     private final AtomicBoolean shutdown = new AtomicBoolean(false);
-    private final ChannelConnector<ServiceStub, ServiceBlockingStub> connector;
+    private final ChannelConnector connector;
     private final Cache cache;
     private final ResolveStrategy strategy;
     private final FlagdOptions options;
@@ -83,7 +84,7 @@ public RpcResolver(
         this.strategy = ResolveFactory.getStrategy(options);
         this.options = options;
         incomingQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
-        this.connector = new ChannelConnector<>(options, onProviderEvent);
+        this.connector = new ChannelConnector(options, onProviderEvent, ChannelBuilder.nettyChannel(options));
         this.onProviderEvent = onProviderEvent;
         this.stub = ServiceGrpc.newStub(this.connector.getChannel()).withWaitForReady();
         this.blockingStub =
@@ -97,7 +98,7 @@ protected RpcResolver(
             final Consumer<FlagdProviderEvent> onProviderEvent,
             ServiceStub mockStub,
             ServiceBlockingStub mockBlockingStub,
-            ChannelConnector<ServiceStub, ServiceBlockingStub> connector) {
+            ChannelConnector connector) {
         this.cache = cache;
         this.strategy = ResolveFactory.getStrategy(options);
         this.options = options;

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

@@ -94,8 +94,7 @@ void whenShuttingDownGrpcConnectorConsumerReceivesDisconnectedEvent() throws Exc
             sync.countDown();
         };
 
-        ChannelConnector<ServiceGrpc.ServiceStub, ServiceGrpc.ServiceBlockingStub> instance =
-                new ChannelConnector<>(FlagdOptions.builder().build(), testConsumer, testChannel);
+        ChannelConnector instance = new ChannelConnector(FlagdOptions.builder().build(), testConsumer, testChannel);
 
         instance.initialize();
         // when shutting grpc connector
@@ -122,8 +121,7 @@ void testMonitorChannelState(ConnectivityState state) throws Exception {
 
         Consumer<FlagdProviderEvent> testConsumer = spy(Consumer.class);
 
-        ChannelConnector<ServiceGrpc.ServiceStub, ServiceGrpc.ServiceBlockingStub> instance =
-                new ChannelConnector<>(FlagdOptions.builder().build(), testConsumer, channel);
+        ChannelConnector instance = new ChannelConnector(FlagdOptions.builder().build(), testConsumer, channel);
 
         instance.initialize();
 

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

@@ -32,7 +32,7 @@
 import org.mockito.stubbing.Answer;
 
 class SyncStreamQueueSourceTest {
-    private ChannelConnector<FlagSyncServiceStub, FlagSyncServiceBlockingStub> mockConnector;
+    private ChannelConnector mockConnector;
     private FlagSyncServiceBlockingStub blockingStub;
     private FlagSyncServiceStub stub;
     private StreamObserver<SyncFlagsResponse> observer;

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/rpc/RpcResolverTest.java

@@ -27,7 +27,7 @@
 import org.mockito.stubbing.Answer;
 
 class RpcResolverTest {
-    private ChannelConnector<ServiceStub, ServiceBlockingStub> mockConnector;
+    private ChannelConnector mockConnector;
     private ServiceBlockingStub blockingStub;
     private ServiceStub stub;
     private QueueingStreamObserver<EventStreamResponse> observer;