Class GrpcFutures

java.lang.Object
ai.pipestream.module.pipelineprobe.grpc.GrpcFutures
public final class GrpcFutures extends Object
Tiny StreamObserver-to-CompletableFuture adapter for unary gRPC calls. Drop-in replacement for the Mutiny stub pattern that was silently eating transport-level failures and hanging the test orchestrator.

Why this exists: Mutiny gRPC stubs (the MutinyXxxGrpc::newMutinyStub factories) wrap a ClientCall in a Uni. The Mutiny adapter has a known failure mode where transport-level terminal events from the underlying ClientCall.Listener don't always materialize into the Uni's onFailure path — especially when the HTTP/2 state machine fires events in an order the adapter doesn't expect. The Uni stays subscribed forever, no callback lands, no thread parks, and the whole test orchestrator stalls with no log line.

Plain async stubs do not have this problem. A StreamObserver receives the gRPC framework's terminal events directly: onNext for the response, onError for any transport or status failure, onCompleted when the call closes. Nothing in between to lose events. The channel's HTTP/2 keepalive (keepAliveTime + keepAliveTimeout on the channel builder) detects dead transports and fires onError with UNAVAILABLE.

Usage: 

{@literal @}GrpcClient("my-service") MyServiceGrpc.MyServiceStub stub;
...
CompletableFuture<MyResponse> future = GrpcFutures.unary(
    observer -> stub.myRpc(request, observer));
MyResponse response = future.get();   // blocks the calling (virtual) thread

This is the same shape the kafka-sidecar uses for its outbound RPCs and matches the gRPC documentation's recommended async-stub usage. No application-level deadlines layered on; channel keepalive handles transport-failure detection. Use deadlines only when you have a real application-level "must complete in N seconds" requirement.

  • Method Details

    • unary

      public static <T> CompletableFuture<T> unary(Consumer<io.grpc.stub.StreamObserver<T>> call)
      Invokes a unary gRPC call and returns a CompletableFuture that completes with the single response, fails with the gRPC status, or fails with IllegalStateException if the call closes without producing a response.
      Type Parameters:
      T - the gRPC response message type produced by the call
      Parameters:
      call - function that takes a response StreamObserver and initiates the gRPC call (e.g. observer -> stub.rpc(req, observer))
      Returns:
      a future completing with the single response, failing with the gRPC status on transport/status error, or failing with IllegalStateException if the call closes without a response