Implement header-based selector for sync stream connection

Co-authored-by: aepfli <9987394+aepfli@users.noreply.github.com>

Author: 2 people

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

@@ -3,7 +3,14 @@
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
 import dev.openfeature.contrib.providers.flagd.resolver.common.nameresolvers.EnvoyResolverProvider;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ForwardingClientCall;
 import io.grpc.ManagedChannel;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
 import io.grpc.NameResolverRegistry;
 import io.grpc.Status.Code;
 import io.grpc.netty.GrpcSslContexts;
@@ -94,14 +101,19 @@ public static ManagedChannel nettyChannel(final FlagdOptions options) {
             if (!Epoll.isAvailable()) {
                 throw new IllegalStateException("unix socket cannot be used", Epoll.unavailabilityCause());
             }
-            return NettyChannelBuilder.forAddress(new DomainSocketAddress(options.getSocketPath()))
+            var channelBuilder = NettyChannelBuilder.forAddress(new DomainSocketAddress(options.getSocketPath()))
                     .keepAliveTime(keepAliveMs, TimeUnit.MILLISECONDS)
                     .eventLoopGroup(new MultiThreadIoEventLoopGroup(EpollIoHandler.newFactory()))
                     .channelType(EpollDomainSocketChannel.class)
                     .usePlaintext()
                     .defaultServiceConfig(buildRetryPolicy(options))
-                    .enableRetry()
-                    .build();
+                    .enableRetry();
+
+            // add header-based selector interceptor if selector is provided
+            if (options.getSelector() != null) {
+                channelBuilder.intercept(createSelectorInterceptor(options.getSelector()));
+            }
+            return channelBuilder.build();
         }
 
         // build a TCP socket
@@ -160,6 +172,30 @@ public static ManagedChannel nettyChannel(final FlagdOptions options) {
         }
     }
 
+    /**
+     * Creates a ClientInterceptor that adds the flagd-selector header to gRPC requests.
+     * This is the preferred approach for passing selectors as per flagd issue #1814.
+     *
+     * @param selector the selector value to pass in the header
+     * @return a ClientInterceptor that adds the flagd-selector header
+     */
+    private static ClientInterceptor createSelectorInterceptor(String selector) {
+        return new ClientInterceptor() {
+            @Override
+            public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+                    MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+                return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
+                        next.newCall(method, callOptions)) {
+                    @Override
+                    public void start(Listener<RespT> responseListener, Metadata headers) {
+                        headers.put(Metadata.Key.of("flagd-selector", Metadata.ASCII_STRING_MARSHALLER), selector);
+                        super.start(responseListener, headers);
+                    }
+                };
+            }
+        };
+    }
+
     private static boolean isValidTargetUri(String targetUri) {
         if (targetUri == null) {
             return false;

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

@@ -16,6 +16,13 @@
 import dev.openfeature.flagd.grpc.sync.Sync.SyncFlagsResponse;
 import dev.openfeature.sdk.Awaitable;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ForwardingClientCall;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
@@ -253,6 +260,8 @@ private void syncFlags(SyncStreamObserver streamObserver) {
         }
 
         final SyncFlagsRequest.Builder syncRequest = SyncFlagsRequest.newBuilder();
+        // Selector is now passed via header using ClientInterceptor (see constructor)
+        // Keeping this for backward compatibility with older flagd versions
         if (this.selector != null) {
             syncRequest.setSelector(this.selector);
         }
@@ -266,6 +275,30 @@ private void syncFlags(SyncStreamObserver streamObserver) {
         streamObserver.done.await();
     }
 
+    /**
+     * Creates a ClientInterceptor that adds the flagd-selector header to gRPC requests.
+     * This is the preferred approach for passing selectors as per flagd issue #1814.
+     *
+     * @param selector the selector value to pass in the header
+     * @return a ClientInterceptor that adds the flagd-selector header
+     */
+    private static ClientInterceptor createSelectorInterceptor(String selector) {
+        return new ClientInterceptor() {
+            @Override
+            public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+                    MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+                return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
+                        next.newCall(method, callOptions)) {
+                    @Override
+                    public void start(Listener<RespT> responseListener, Metadata headers) {
+                        headers.put(Metadata.Key.of("flagd-selector", Metadata.ASCII_STRING_MARSHALLER), selector);
+                        super.start(responseListener, headers);
+                    }
+                };
+            }
+        };
+    }
+
     private void enqueueError(String message) {
         enqueueError(outgoingQueue, message);
     }