fix: RPC mode does not honor timeout

During the refactoring of the connection the timeout for
unary calls got lost. This adds the timeout back into the
mix.

TODO:
- [ ] adapt gherkin tests for contextEnrichment to do handle stale seperately for file and inprocess

Signed-off-by: Simon Schrottner <[email protected]>

Author: aepfli

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

@@ -32,7 +32,7 @@ public class GrpcConnector<T extends AbstractStub<T>, K extends AbstractBlocking
     /**
      * The blocking service stub for making blocking GRPC calls.
      */
-    private final K blockingStub;
+    private final Function<ManagedChannel, K> blockingStubFunction;
 
     /**
      * The GRPC managed channel for managing the underlying GRPC connection.
@@ -59,6 +59,8 @@ public class GrpcConnector<T extends AbstractStub<T>, K extends AbstractBlocking
      */
     private final Consumer<T> streamObserver;
 
+    private final FlagdOptions options;
+
     /**
      * Indicates whether the connector is currently connected to the GRPC service.
      */
@@ -85,11 +87,12 @@ public GrpcConnector(
 
         this.channel = channel;
         this.serviceStub = stub.apply(channel).withWaitForReady();
-        this.blockingStub = blockingStub.apply(channel).withWaitForReady();
+        this.blockingStubFunction = blockingStub;
         this.deadline = options.getDeadline();
         this.streamDeadlineMs = options.getStreamDeadlineMs();
         this.onConnectionEvent = onConnectionEvent;
         this.streamObserver = eventStreamObserver;
+        this.options = options;
     }
 
     /**
@@ -126,7 +129,7 @@ public void initialize() throws Exception {
      * @return the blocking service stub
      */
     public K getResolver() {
-        return blockingStub;
+        return blockingStubFunction.apply(channel).withWaitForReady();
     }
 
     /**

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/grpc/GrpcResolver.java

@@ -36,6 +36,7 @@
 import io.grpc.Status.Code;
 import io.grpc.StatusRuntimeException;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
@@ -50,6 +51,7 @@ public final class GrpcResolver implements Resolver {
     private final GrpcConnector<ServiceGrpc.ServiceStub, ServiceGrpc.ServiceBlockingStub> connector;
     private final Cache cache;
     private final ResolveStrategy strategy;
+    private final FlagdOptions options;
 
     /**
      * Resolves flag values using https://buf.build/open-feature/flagd/docs/main:flagd.evaluation.v1.
@@ -63,6 +65,7 @@ public GrpcResolver(
             final FlagdOptions options, final Cache cache, final Consumer<FlagdProviderEvent> onProviderEvent) {
         this.cache = cache;
         this.strategy = ResolveFactory.getStrategy(options);
+        this.options = options;
         this.connector = new GrpcConnector<>(
                 options,
                 ServiceGrpc::newStub,
@@ -108,15 +111,15 @@ public void onError() {
     public ProviderEvaluation<Boolean> booleanEvaluation(String key, Boolean defaultValue, EvaluationContext ctx) {
         ResolveBooleanRequest request = ResolveBooleanRequest.newBuilder().buildPartial();
 
-        return resolve(key, ctx, request, connector.getResolver()::resolveBoolean, null);
+        return resolve(key, ctx, request, getResolver()::resolveBoolean, null);
     }
 
     /**
      * String evaluation from grpc resolver.
      */
     public ProviderEvaluation<String> stringEvaluation(String key, String defaultValue, EvaluationContext ctx) {
         ResolveStringRequest request = ResolveStringRequest.newBuilder().buildPartial();
-        return resolve(key, ctx, request, connector.getResolver()::resolveString, null);
+        return resolve(key, ctx, request, getResolver()::resolveString, null);
     }
 
     /**
@@ -125,7 +128,7 @@ public ProviderEvaluation<String> stringEvaluation(String key, String defaultVal
     public ProviderEvaluation<Double> doubleEvaluation(String key, Double defaultValue, EvaluationContext ctx) {
         ResolveFloatRequest request = ResolveFloatRequest.newBuilder().buildPartial();
 
-        return resolve(key, ctx, request, connector.getResolver()::resolveFloat, null);
+        return resolve(key, ctx, request, getResolver()::resolveFloat, null);
     }
 
     /**
@@ -135,8 +138,11 @@ public ProviderEvaluation<Integer> integerEvaluation(String key, Integer default
 
         ResolveIntRequest request = ResolveIntRequest.newBuilder().buildPartial();
 
-        return resolve(
-                key, ctx, request, connector.getResolver()::resolveInt, (Object value) -> ((Long) value).intValue());
+        return resolve(key, ctx, request, getResolver()::resolveInt, (Object value) -> ((Long) value).intValue());
+    }
+
+    private ServiceGrpc.ServiceBlockingStub getResolver() {
+        return connector.getResolver().withDeadlineAfter(options.getDeadline(), TimeUnit.MILLISECONDS);
     }
 
     /**
@@ -150,7 +156,7 @@ public ProviderEvaluation<Value> objectEvaluation(String key, Value defaultValue
                 key,
                 ctx,
                 request,
-                connector.getResolver()::resolveObject,
+                getResolver()::resolveObject,
                 (Object value) -> convertObjectResponse((Struct) value));
     }