From 335d6f0320cf9c5fc3255733dd858ad7b9a356fd Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 24 Sep 2021 14:14:35 +0200 Subject: [PATCH] GH-744 Add initial support for client-side streaming to gRPC --- .../grpc/GrpcServerMessageHandler.java | 116 +++++++++++++++--- .../cloud/function/grpc/GrpcUtils.java | 79 +++++++++++- .../function/grpc/GrpcInteractionTests.java | 39 +++++- .../src/test/resources/application.properties | 2 +- 4 files changed, 210 insertions(+), 26 deletions(-) diff --git a/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServerMessageHandler.java b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServerMessageHandler.java index 0b5dbd2c9..dc5960cd5 100644 --- a/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServerMessageHandler.java +++ b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServerMessageHandler.java @@ -33,6 +33,10 @@ package org.springframework.cloud.function.grpc; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import io.grpc.Status; @@ -40,6 +44,7 @@ import io.grpc.stub.ServerCallStreamObserver; import io.grpc.stub.StreamObserver; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Sinks; import reactor.core.publisher.Sinks.Many; @@ -86,13 +91,79 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { // StreamObserver responseObserver) { // // } -// -// @Override -// public StreamObserver clientStream( -// StreamObserver responseObserver) { -// return null; -// } -// + + @SuppressWarnings("unchecked") + @Override + public StreamObserver clientStream(StreamObserver responseObserver) { + ServerCallStreamObserver serverCallStreamObserver = (ServerCallStreamObserver) responseObserver; + serverCallStreamObserver.disableAutoInboundFlowControl(); + + AtomicBoolean wasReady = new AtomicBoolean(false); + serverCallStreamObserver.setOnReadyHandler(() -> { + if (serverCallStreamObserver.isReady() && !wasReady.get()) { + wasReady.set(true); + logger.info("gRPC Server receiving stream is ready."); + serverCallStreamObserver.request(1); + } + }); + + + if (!this.function.isInputTypePublisher()) { + throw new UnsupportedOperationException("The client streaming is " + + "not supported for functions that accept non-Publisher: " + + this.function); + } + else if (this.function.isOutputTypePublisher()) { + throw new UnsupportedOperationException("The client streaming is " + + "not supported for functions that return Publisher: " + + this.function); + } + else { + Many> inputStream = Sinks.many().unicast().onBackpressureBuffer(); + Flux> inputStreamFlux = inputStream.asFlux(); + + LinkedBlockingQueue> resultRef = new LinkedBlockingQueue<>(1); + ExecutorService executor = Executors.newSingleThreadExecutor(); + executor.execute(() -> { + Message replyMessage = (Message) this.function.apply(inputStreamFlux); + System.out.println(replyMessage); + resultRef.offer(replyMessage); + }); + + return new StreamObserver() { + + @Override + public void onNext(GrpcMessage inputMessage) { + if (logger.isDebugEnabled()) { + logger.debug("gRPC Server receiving: " + inputMessage); + } + + inputStream.tryEmitNext(GrpcUtils.fromGrpcMessage(inputMessage)); + serverCallStreamObserver.request(1); + } + + @Override + public void onError(Throwable t) { + t.printStackTrace(); + responseObserver.onCompleted(); + } + + @Override + public void onCompleted() { + logger.info("gRPC Server has finished receiving data."); + inputStream.tryEmitComplete(); + try { + responseObserver.onNext(GrpcUtils.toGrpcMessage(resultRef.poll(5000, TimeUnit.MILLISECONDS))); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + responseObserver.onCompleted(); + } + }; + } + } + @Override public StreamObserver biStream(StreamObserver responseObserver) { ServerCallStreamObserver serverCallStreamObserver = (ServerCallStreamObserver) responseObserver; @@ -102,7 +173,7 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { serverCallStreamObserver.setOnReadyHandler(() -> { if (serverCallStreamObserver.isReady() && !wasReady.get()) { wasReady.set(true); - logger.info("Server stream is ready"); + logger.info("gRPC Server receiving stream is ready."); serverCallStreamObserver.request(1); } }); @@ -166,7 +237,7 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { @Override public void onCompleted() { - logger.info("Server Stream is complete"); + logger.info("gRPC Server has finished receiving data."); responseObserver.onCompleted(); } }; @@ -175,21 +246,28 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { @SuppressWarnings("unchecked") private StreamObserver biStreamReactive(StreamObserver responseObserver, ServerCallStreamObserver serverCallStreamObserver) { - Many> sink = Sinks.many().unicast().onBackpressureBuffer(); - Flux> flux = sink.asFlux(); + Many> inputStream = Sinks.many().unicast().onBackpressureBuffer(); + Flux> inputStreamFlux = inputStream.asFlux(); - Flux> connectedFlux = (Flux>) function.apply(flux); + Publisher> outputPublisher = (Publisher>) this.function.apply(inputStreamFlux); - connectedFlux.subscribe(functionResult -> { - GrpcMessage reply = GrpcUtils.toGrpcMessage(functionResult); - responseObserver.onNext(reply); + Flux.from(outputPublisher).subscribe(functionResult -> { + GrpcMessage outputMessage = GrpcUtils.toGrpcMessage(functionResult); + if (logger.isDebugEnabled()) { + logger.debug("gRPC Server replying: " + outputMessage); + } + responseObserver.onNext(outputMessage); }); return new StreamObserver() { @Override - public void onNext(GrpcMessage value) { - sink.tryEmitNext(GrpcUtils.fromGrpcMessage(value)); + public void onNext(GrpcMessage inputMessage) { + if (logger.isDebugEnabled()) { + logger.debug("gRPC Server receiving: " + inputMessage); + } + + inputStream.tryEmitNext(GrpcUtils.fromGrpcMessage(inputMessage)); serverCallStreamObserver.request(1); } @@ -201,8 +279,8 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { @Override public void onCompleted() { - logger.info("Server stream is complete"); - sink.tryEmitComplete(); + logger.info("gRPC Server has finished receiving data."); + inputStream.tryEmitComplete(); responseObserver.onCompleted(); } }; diff --git a/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcUtils.java b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcUtils.java index ad5195e97..f0d7367e0 100644 --- a/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcUtils.java +++ b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcUtils.java @@ -18,12 +18,15 @@ package org.springframework.cloud.function.grpc; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import com.google.protobuf.ByteString; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import io.grpc.stub.ClientCallStreamObserver; import io.grpc.stub.ClientResponseObserver; +import io.grpc.stub.StreamObserver; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import reactor.core.publisher.Flux; @@ -91,7 +94,7 @@ final class GrpcUtils { * They are completely independent where one may end before the other. * * @param inputStream {@code FluxMessage>} representing input stream. - * @return {@code FluxMessage>} representing output stream + * @return {@code Flux>} representing output stream */ public static Flux> biStreaming(Flux> inputStream) { return biStreaming("localhost", FunctionGrpcProperties.GRPC_PORT, inputStream); @@ -105,7 +108,7 @@ final class GrpcUtils { * @param host gRPC server host name * @param port gRPC server port * @param inputStream {@code FluxMessage>} representing input stream - * @return {@code FluxMessage>} representing output stream + * @return {@code Flux>} representing output stream */ public static Flux> biStreaming(String host, int port, Flux> inputStream) { ManagedChannel channel = ManagedChannelBuilder @@ -125,6 +128,77 @@ final class GrpcUtils { }); } + /** + * Utility method to support client-side streaming interaction. Will connect to gRPC server using default host/port, + * otherwise use {@link #clientStream(String, int, Flux)} method. + * + * @param inputStream {@code FluxMessage>} representing input stream. + * @return {@code Message} representing output + */ + public static Message clientStream(Flux> inputStream) { + return clientStream("localhost", FunctionGrpcProperties.GRPC_PORT, inputStream); + } + + /** + * Utility method to support client-side streaming interaction. + * + * @param host gRPC server host name + * @param port gRPC server port + * @param inputStream {@code FluxMessage>} representing input stream + * @return {@code Message} representing output + */ + public static Message clientStream(String host, int port, Flux> inputStream) { + ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port) + .usePlaintext().build(); + + LinkedBlockingQueue> resultRef = new LinkedBlockingQueue<>(1); + StreamObserver responseObserver = new StreamObserver() { + @Override + public void onNext(GrpcMessage result) { + if (logger.isDebugEnabled()) { + logger.debug("Client received reply: " + result); + } + resultRef.offer(GrpcUtils.fromGrpcMessage(result)); + } + + @Override + public void onError(Throwable t) { + t.printStackTrace(); + } + + @Override + public void onCompleted() { + logger.info("Client completed"); + } + }; + + MessagingServiceGrpc.MessagingServiceStub asyncStub = MessagingServiceGrpc.newStub(channel); + + StreamObserver requestObserver = asyncStub.clientStream(responseObserver); + + inputStream.doOnNext(message -> { + if (logger.isDebugEnabled()) { + logger.debug("Client sending: " + message); + } + try { + requestObserver.onNext(GrpcUtils.toGrpcMessage(message)); + } + catch (Exception e) { + requestObserver.onError(e); + } + }).doOnComplete(() -> { + requestObserver.onCompleted(); + }).subscribe(); + + try { + return resultRef.poll(5000, TimeUnit.MILLISECONDS); + } + catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(ie); + } + } + private static ClientResponseObserver clientResponseObserver(Flux> inputStream, Many> sink) { return new ClientResponseObserver() { @@ -155,7 +229,6 @@ final class GrpcUtils { @Override public void onNext(GrpcMessage message) { - System.out.println("RECEIVED: " + message); if (logger.isDebugEnabled()) { logger.debug("Receiving message: " + message); } diff --git a/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java b/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java index 5b65b3916..38bb37438 100644 --- a/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java +++ b/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java @@ -19,6 +19,7 @@ package org.springframework.cloud.function.grpc; import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.Random; import java.util.function.Function; import org.junit.jupiter.api.Test; @@ -129,7 +130,32 @@ public class GrpcInteractionTests { } @Test - public void testStreamInStringOut() { + public void testClientStreaming() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + SampleConfiguration.class).web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.function.definition=streamInStringOut", + "--spring.cloud.function.grpc.port=" + + FunctionGrpcProperties.GRPC_PORT, + "--spring.cloud.function.grpc.mode=server")) { + + List> messages = new ArrayList<>(); + messages.add(MessageBuilder.withPayload("\"Ricky\"".getBytes()).setHeader("foo", "bar") + .build()); + messages.add(MessageBuilder.withPayload("\"Julien\"".getBytes()).setHeader("foo", "bar") + .build()); + messages.add(MessageBuilder.withPayload("\"Bubbles\"".getBytes()).setHeader("foo", "bar") + .build()); + + Message reply = + GrpcUtils.clientStream("localhost", FunctionGrpcProperties.GRPC_PORT, Flux.fromIterable(messages)); + + assertThat(reply.getPayload()).isEqualTo("[Ricky, Julien, Bubbles]".getBytes()); + } + } + + @Test + public void testBiStreamStreamInStringOutFailure() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( SampleConfiguration.class).web(WebApplicationType.NONE).run( "--spring.jmx.enabled=false", @@ -163,7 +189,7 @@ public class GrpcInteractionTests { } @Test - public void testStringInStreamOut() { + public void testBiStreamStringInStreamOutFailure() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( SampleConfiguration.class).web(WebApplicationType.NONE).run( "--spring.jmx.enabled=false", @@ -211,7 +237,14 @@ public class GrpcInteractionTests { @Bean public Function, String> streamInStringOut() { - return flux -> "hello"; + return flux -> flux.doOnNext(v -> { + try { + Thread.sleep(new Random().nextInt(2000)); // artificial delay + } + catch (Exception e) { + // ignore + } + }).collectList().block().toString(); } @Bean diff --git a/spring-cloud-function-grpc/src/test/resources/application.properties b/spring-cloud-function-grpc/src/test/resources/application.properties index 8b1378917..1aab44a8b 100644 --- a/spring-cloud-function-grpc/src/test/resources/application.properties +++ b/spring-cloud-function-grpc/src/test/resources/application.properties @@ -1 +1 @@ - +logging.level.org.springframework.cloud.function.grpc=DEBUG