diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 93a2230b9a4..857e08ecd96 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -934,6 +934,10 @@ boolean isSidecarReady() { return observer != null && observer.isReady(); } + private boolean isUpstreamReady() { + return method.getType().clientSendsOneMessage() || super.isReady(); + } + @Override public boolean isReady() { if (passThroughMode.get()) { @@ -1193,7 +1197,7 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) { boolean sendImmediately = false; synchronized (streamLock) { sidestreamToUpstreamWindow -= body.size(); - if (pendingUpstreamBodyMessages.isEmpty() && super.isReady()) { + if (pendingUpstreamBodyMessages.isEmpty() && isUpstreamReady()) { sendImmediately = true; accumulatedWindowUpdateSidestreamToUpstream += body.size(); } else { @@ -1307,7 +1311,7 @@ void drainPendingUpstreamBodyMessages() { ByteString body = null; boolean triggerHalfClose = false; synchronized (streamLock) { - if (!pendingUpstreamBodyMessages.isEmpty() && super.isReady()) { + if (!pendingUpstreamBodyMessages.isEmpty() && isUpstreamReady()) { body = pendingUpstreamBodyMessages.poll(); accumulatedWindowUpdateSidestreamToUpstream += body.size(); if (pendingUpstreamBodyMessages.isEmpty() @@ -1366,16 +1370,16 @@ private void drainPendingDrainingMessages() { boolean triggerHalfClose = false; synchronized (streamLock) { - if (!pendingUpstreamBodyMessages.isEmpty() && super.isReady()) { + if (!pendingUpstreamBodyMessages.isEmpty() && isUpstreamReady()) { msg = pendingUpstreamBodyMessages.poll(); isMutated = true; } else if (pendingUpstreamBodyMessages.isEmpty() - && !pendingRequestBodyMessages.isEmpty() && super.isReady()) { + && !pendingRequestBodyMessages.isEmpty() && isUpstreamReady()) { msg = pendingRequestBodyMessages.poll(); isMutated = true; } else if (pendingUpstreamBodyMessages.isEmpty() && pendingRequestBodyMessages.isEmpty() - && !pendingDrainingMessages.isEmpty() && super.isReady()) { + && !pendingDrainingMessages.isEmpty() && isUpstreamReady()) { msg = pendingDrainingMessages.poll(); isMutated = false; } diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index 48f0931cbc5..19ae0940bd7 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -50,6 +50,7 @@ import io.grpc.Context; import io.grpc.Deadline; import io.grpc.ForwardingClientCall.SimpleForwardingClientCall; +import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener; import io.grpc.ManagedChannel; import io.grpc.Metadata; import io.grpc.MethodDescriptor; @@ -154,6 +155,14 @@ public class ExternalProcessorClientInterceptorTest { .setResponseMarshaller(new StringMarshaller()) .build(); + private static final MethodDescriptor METHOD_SERVER_STREAMING = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.SERVER_STREAMING) + .setFullMethodName("test.TestService/ServerStreaming") + .setRequestMarshaller(new StringMarshaller()) + .setResponseMarshaller(new StringMarshaller()) + .build(); + private static final MethodDescriptor METHOD_BIDI_STREAMING = MethodDescriptor.newBuilder() .setType(MethodDescriptor.MethodType.BIDI_STREAMING) @@ -22397,6 +22406,821 @@ public void cancel(String message, Throwable cause) { shutdownAndAwaitTermination(executor); } + // --- Category 30: Unary and Server-Streaming Outbound Flow Control --- + + @Test + @SuppressWarnings("unchecked") + public void unaryCall_whenUpstreamNotReady_messageAndHalfCloseForwardedImmediately() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setResponseBodyMode(ProcessingMode.BodySendMode.NONE) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } else if (request.hasRequestBody()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("mutated-ping")) + .setEndOfStream(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } else if (request.hasResponseHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final AtomicReference receivedServerRequest = new AtomicReference<>(); + final AtomicBoolean serverReceivedHalfClose = new AtomicBoolean(false); + final CountDownLatch serverDoneLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall((request, responseObserver) -> { + receivedServerRequest.set(request); + serverReceivedHalfClose.set(true); + responseObserver.onNext("Hello " + request); + responseObserver.onCompleted(); + serverDoneLatch.countDown(); + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + // Interceptor simulating DelayedStream / stream allocation delay where isReady() returns false + // and onReady() is suppressed (matching ClientCallImpl behavior for clientSendsOneMessage()). + ClientInterceptor unreadyUpstreamInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public boolean isReady() { + return false; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + super.start(new SimpleForwardingClientCallListener(responseListener) { + @Override + public void onReady() { + // Suppress onReady matching ClientCallImpl behavior for clientSendsOneMessage() + } + }, headers); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(unreadyUpstreamInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_SAY_HELLO, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + final AtomicReference clientReceivedResponse = new AtomicReference<>(); + final AtomicReference clientReceivedStatus = new AtomicReference<>(); + final CountDownLatch clientDoneLatch = new CountDownLatch(1); + + proxyCall.start(new ClientCall.Listener() { + @Override + public void onMessage(String message) { + clientReceivedResponse.set(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + clientReceivedStatus.set(status); + clientDoneLatch.countDown(); + } + }, new Metadata()); + + proxyCall.request(1); + proxyCall.sendMessage("ping"); + proxyCall.halfClose(); + + assertThat(serverDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + assertThat(receivedServerRequest.get()).isEqualTo("mutated-ping"); + assertThat(serverReceivedHalfClose.get()).isTrue(); + assertThat(clientReceivedResponse.get()).isEqualTo("Hello mutated-ping"); + assertThat(clientReceivedStatus.get().getCode()).isEqualTo(Status.Code.OK); + + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void unaryCall_whenExtProcSendsEndOfStreamWithoutMessage_whenUpstreamNotReady_succeeds() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setResponseBodyMode(ProcessingMode.BodySendMode.NONE) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } else if (request.hasRequestBody()) { + // Response 1: body message without end_of_stream + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("chunk-ping")) + .setEndOfStream(false) + .build()) + .build()) + .build()) + .build()) + .build()); + // Response 2: end_of_stream without message + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } else if (request.hasResponseHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final List receivedServerRequests = new ArrayList<>(); + final CountDownLatch serverDoneLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall((request, responseObserver) -> { + receivedServerRequests.add(request); + responseObserver.onNext("Hello " + request); + responseObserver.onCompleted(); + serverDoneLatch.countDown(); + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + ClientInterceptor unreadyUpstreamInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public boolean isReady() { + return false; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + super.start(new SimpleForwardingClientCallListener(responseListener) { + @Override + public void onReady() {} + }, headers); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(unreadyUpstreamInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_SAY_HELLO, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + final CountDownLatch clientDoneLatch = new CountDownLatch(1); + final AtomicReference clientStatus = new AtomicReference<>(); + + proxyCall.start(new ClientCall.Listener() { + @Override + public void onClose(Status status, Metadata trailers) { + clientStatus.set(status); + clientDoneLatch.countDown(); + } + }, new Metadata()); + + proxyCall.request(1); + proxyCall.sendMessage("ping"); + proxyCall.halfClose(); + + assertThat(serverDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(receivedServerRequests).containsExactly("chunk-ping"); + assertThat(clientStatus.get().getCode()).isEqualTo(Status.Code.OK); + + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void unaryCall_whenExtProcDraining_whenUpstreamNotReady_drainingMessagesFlushed() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setResponseBodyMode(ProcessingMode.BodySendMode.NONE) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .setRequestDrain(true) + .build()); + } else if (request.hasResponseHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final AtomicReference receivedServerRequest = new AtomicReference<>(); + final CountDownLatch serverDoneLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall((request, responseObserver) -> { + receivedServerRequest.set(request); + responseObserver.onNext("Hello " + request); + responseObserver.onCompleted(); + serverDoneLatch.countDown(); + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + ClientInterceptor unreadyUpstreamInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public boolean isReady() { + return false; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + super.start(new SimpleForwardingClientCallListener(responseListener) { + @Override + public void onReady() {} + }, headers); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(unreadyUpstreamInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_SAY_HELLO, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + final AtomicReference clientResponse = new AtomicReference<>(); + final AtomicReference clientStatus = new AtomicReference<>(); + final CountDownLatch clientDoneLatch = new CountDownLatch(1); + + proxyCall.start(new ClientCall.Listener() { + @Override + public void onMessage(String message) { + clientResponse.set(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + clientStatus.set(status); + clientDoneLatch.countDown(); + } + }, new Metadata()); + + proxyCall.request(1); + proxyCall.sendMessage("unmutated-ping"); + proxyCall.halfClose(); + + assertThat(serverDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(receivedServerRequest.get()).isEqualTo("unmutated-ping"); + assertThat(clientResponse.get()).isEqualTo("Hello unmutated-ping"); + assertThat(clientStatus.get().getCode()).isEqualTo(Status.Code.OK); + + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void serverStreamingCall_whenUpstreamNotReady_messageAndHalfCloseForwarded() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setResponseBodyMode(ProcessingMode.BodySendMode.NONE) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } else if (request.hasRequestBody()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("stream-query")) + .setEndOfStream(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } else if (request.hasResponseHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final AtomicReference receivedServerQuery = new AtomicReference<>(); + final CountDownLatch serverDoneLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SERVER_STREAMING, + ServerCalls.asyncServerStreamingCall((request, responseObserver) -> { + receivedServerQuery.set(request); + responseObserver.onNext("Item 1 for " + request); + responseObserver.onNext("Item 2 for " + request); + responseObserver.onCompleted(); + serverDoneLatch.countDown(); + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + ClientInterceptor unreadyUpstreamInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public boolean isReady() { + return false; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + super.start(new SimpleForwardingClientCallListener(responseListener) { + @Override + public void onReady() {} + }, headers); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(unreadyUpstreamInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_SERVER_STREAMING, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + final List clientItems = new ArrayList<>(); + final AtomicReference clientStatus = new AtomicReference<>(); + final CountDownLatch clientDoneLatch = new CountDownLatch(1); + + proxyCall.start(new ClientCall.Listener() { + @Override + public void onMessage(String message) { + clientItems.add(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + clientStatus.set(status); + clientDoneLatch.countDown(); + } + }, new Metadata()); + + proxyCall.request(10); + proxyCall.sendMessage("query"); + proxyCall.halfClose(); + + assertThat(serverDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(receivedServerQuery.get()).isEqualTo("stream-query"); + assertThat(clientItems) + .containsExactly("Item 1 for stream-query", "Item 2 for stream-query"); + assertThat(clientStatus.get().getCode()).isEqualTo(Status.Code.OK); + + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void clientStreamingCall_whenUpstreamNotReady_messageGatedUntilOnReady() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setResponseBodyMode(ProcessingMode.BodySendMode.NONE) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } else if (request.hasRequestBody()) { + boolean eosWithoutMsg = + request.getRequestBody().getEndOfStreamWithoutMessage(); + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(request.getRequestBody().getBody()) + .setEndOfStream(request.getRequestBody().getEndOfStream()) + .setEndOfStreamWithoutMessage(eosWithoutMsg) + .build()) + .build()) + .build()) + .build()) + .build()); + } else if (request.hasResponseHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final List receivedServerRequests = new ArrayList<>(); + final CountDownLatch serverReceivedFirstMessageLatch = new CountDownLatch(1); + final CountDownLatch serverDoneLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_CLIENT_STREAMING, + ServerCalls.asyncClientStreamingCall(responseObserver -> { + return new StreamObserver() { + @Override + public void onNext(String value) { + receivedServerRequests.add(value); + serverReceivedFirstMessageLatch.countDown(); + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onNext("Total: " + receivedServerRequests.size()); + responseObserver.onCompleted(); + serverDoneLatch.countDown(); + } + }; + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + final AtomicBoolean upstreamIsReady = new AtomicBoolean(false); + final AtomicReference> capturedListener = new AtomicReference<>(); + + ClientInterceptor controllableInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public boolean isReady() { + return upstreamIsReady.get(); + } + + @Override + public void start(Listener responseListener, Metadata headers) { + capturedListener.set(responseListener); + super.start(responseListener, headers); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(controllableInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_CLIENT_STREAMING, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + final AtomicReference clientResponse = new AtomicReference<>(); + final AtomicReference clientStatus = new AtomicReference<>(); + final CountDownLatch clientDoneLatch = new CountDownLatch(1); + + proxyCall.start(new ClientCall.Listener() { + @Override + public void onMessage(String message) { + clientResponse.set(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + clientStatus.set(status); + clientDoneLatch.countDown(); + } + }, new Metadata()); + + proxyCall.request(1); + // Send message while upstream is NOT ready + proxyCall.sendMessage("msg1"); + + // Give ext-proc time to process and return BodyResponse + // Upstream should NOT have received it yet because isReady is false for client streaming + assertThat(serverReceivedFirstMessageLatch.await(200, TimeUnit.MILLISECONDS)).isFalse(); + assertThat(receivedServerRequests).isEmpty(); + + // Now make upstream ready and trigger onReady() + upstreamIsReady.set(true); + capturedListener.get().onReady(); + + // Now msg1 should be flushed to server + assertThat(serverReceivedFirstMessageLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(receivedServerRequests).containsExactly("msg1"); + + proxyCall.halfClose(); + assertThat(serverDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientResponse.get()).isEqualTo("Total: 1"); + assertThat(clientStatus.get().getCode()).isEqualTo(Status.Code.OK); + + channelManager.close(); + } + private static List filterClientRequests(List requests) { List clientRequests = new ArrayList<>(); for (ProcessingRequest r : requests) {