diff --git a/spring-cloud-function-grpc/README.md b/spring-cloud-function-grpc/README.md index 7991ad4f2..e66f2518c 100644 --- a/spring-cloud-function-grpc/README.md +++ b/spring-cloud-function-grpc/README.md @@ -12,13 +12,14 @@ As with all other Spring-boot based frameworks all you need to do is add `spring ``` ### Programming model + +#### Two operation modes (client/server) Spring Cloud Function gRPC support provides two modes of operation - _client_ and _server_. In other words when you add `spring-cloud-function-grpc` dependency to your POM you may or may not want the gRPC server as you may only be interested in client-side utilities to invoke a function exposed via gRPC server running on some host/port. To support these two modes Spring Cloud Function provides `spring.cloud.function.grpc.server` which defaults to `true`. -This means that the default mode of operation is _server_, since the core imtention of gRPC support is to expose user Function via gRPC. However, if you're only inteersted in using client-side utilities (e.g., `GrpcUtils` to help to invoke a function or convert `GrpcMessage` to Spring `Message` and vice versa), you can set this property to `false`. -Hoever if you intention is to - +This means that the default mode of operation is _server_, since the core intention of our current gRPC support is to expose user Functions via gRPC. However, if you're only inteersted in using client-side utilities (e.g., `GrpcUtils` to help to invoke a function or convert `GrpcMessage` to Spring `Message` and vice versa), you can set this property to `false`. +#### Core Data and Service At the center of gRPC and Spring Cloud Function integration is a canonical protobuff structure - `GrpcMessage`. It is modeled after Spring [Message](https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/messaging/Message.html). ``` @@ -44,18 +45,17 @@ service MessagingService { That said, when using Java, you do not need to generate anything, rather identify function definition and send and receive Spring `Messages`. You can get a pretty good idea from this [test case](https://github.com/spring-cloud/spring-cloud-function/blob/82e2583acd7c8aaaf2bc5ec935d486a336e97ae7/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java#L49). -#### 4 Interaction Modes +#### 4 Interaction RPC Modes The gRPC provides 4 interaction modes -* Reques/Repply -* Server-side streaming -* Client-side streaming -* Bi-directional streaming +* Reques/Repply RPC +* Server-side streaming RPC +* Client-side streaming RPC +* Bi-directional streaming RPC Spring Cloud Function provides support for all 4 of them. - -##### Request Reply +##### Request Reply RPC The most straight forward interaction mode is _Request/Reply_. Suppose you have a function @@ -68,10 +68,125 @@ public static class SampleConfiguration { } } ``` -You can invoke it using utility method(s) provided in `GrpcUtils` class +After identifying this function via `spring.cloud.function.definition` property (see example [here](https://github.com/spring-cloud/spring-cloud-function/blob/ded02fec0a6d3d66b8ec00f99f28be2a4bbec668/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java)), +you can invoke it using utility method(s) provided in `GrpcUtils` class ```java Message message = MessageBuilder.withPayload("\"hello gRPC\"".getBytes()) - .setHeader("foo", "bar") - .build(); + .setHeader("foo", "bar") + .build(); Message reply = GrpcUtils.requestReply(message); -``` \ No newline at end of file +``` + +You can also provide `spring.cloud.function.definition` property via `Message` headers, to support more dynamic cases. +```java +Message message = MessageBuilder.withPayload("\"hello gRPC\"".getBytes()) + .setHeader("foo", "bar") + .setHeader("spring.cloud.function.definition", "reverse") + .build(); +``` + +##### Server-side streaming RPC +The Server-side streaming RPC allows you to reply with the stream of data. + +```java +@EnableAutoConfiguration +public static class SampleConfiguration { + @Bean + public Function> stringInStreamOut() { + return value -> Flux.just(value, value.toUpperCase()); + } +} +``` +After identifying this function via `spring.cloud.function.definition` property (see example [here](https://github.com/spring-cloud/spring-cloud-function/blob/ded02fec0a6d3d66b8ec00f99f28be2a4bbec668/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java)), +you can invoke it using utility method(s) provided in `GrpcUtils` class +```java +Message message = MessageBuilder.withPayload("\"hello gRPC\"".getBytes()).setHeader("foo", "bar").build(); + +Flux> reply = + GrpcUtils.serverStream("localhost", FunctionGrpcProperties.GRPC_PORT, message); + +List> results = reply.collectList().block(Duration.ofSeconds(5)); +``` + +You can see that gRPC stream is mapped to instance of `Flux` from [project reactor](https://projectreactor.io/) + +Similarly to the _request/reply_ you can also provide `spring.cloud.function.definition` property via `Message` headers, to support more dynamic cases. +```java +Message message = MessageBuilder.withPayload("\"hello gRPC\"".getBytes()) + .setHeader("foo", "bar") + .setHeader("spring.cloud.function.definition", "reverse") + .build(); +``` + +##### Client-side streaming RPC +The Client-side streaming RPC allows you to stream input data and receive a single reply. + +```java +@EnableAutoConfiguration +public static class SampleConfiguration { + @Bean + public Function, String> streamInStringOut() { + return flux -> flux.doOnNext(v -> { + try { + // do something useful + Thread.sleep(new Random().nextInt(2000)); // artificial delay + } + catch (Exception e) { + // ignore + } + }).collectList().block().toString(); + } +} +``` +After identifying this function via `spring.cloud.function.definition` property (see example [here](https://github.com/spring-cloud/spring-cloud-function/blob/ded02fec0a6d3d66b8ec00f99f28be2a4bbec668/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java)), +you can invoke it using utility method(s) provided in `GrpcUtils` class +```java +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)); + +``` + +You can see that gRPC stream is mapped to instance of `Flux` from [project reactor](https://projectreactor.io/) + +Unlike the _request/reply_ and _server-side streaming_, you can ONLY pass function definition via property or environment variable. + +##### Bi-Directional streaming RPC +The bi-directional streaming RPC allows you to stream input and output data. + +```java +@EnableAutoConfiguration +public static class SampleConfiguration { + @Bean + public Function, Flux> uppercaseReactive() { + return flux -> flux.map(v -> v.toUpperCase()); + } +} +``` +After identifying this function via `spring.cloud.function.definition` property (see example [here](https://github.com/spring-cloud/spring-cloud-function/blob/ded02fec0a6d3d66b8ec00f99f28be2a4bbec668/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java)), +you can invoke it using utility method(s) provided in `GrpcUtils` class +```java +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()); + +Flux> clientResponseObserver = + GrpcUtils.biStreaming("localhost", FunctionGrpcProperties.GRPC_PORT, Flux.fromIterable(messages)); + +List> results = clientResponseObserver.collectList().block(Duration.ofSeconds(1)); +``` + +You can see that gRPC stream is mapped to instance of `Flux` from [project reactor](https://projectreactor.io/) + +Unlike the _request/reply_ and _server-side streaming_, you can ONLY pass function definition via property or environment variable. 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 eb3bd22ab..aa7cf1dc8 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,7 @@ package org.springframework.cloud.function.grpc; +import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; @@ -53,8 +54,10 @@ import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionProperties; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.grpc.MessagingServiceGrpc.MessagingServiceImplBase; +import org.springframework.context.SmartLifecycle; import org.springframework.messaging.Message; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; /** * @@ -62,23 +65,32 @@ import org.springframework.util.Assert; * @since 3.2 * */ -class GrpcServerMessageHandler extends MessagingServiceImplBase { +class GrpcServerMessageHandler extends MessagingServiceImplBase implements SmartLifecycle { private Log logger = LogFactory.getLog(GrpcServerMessageHandler.class); - private final FunctionInvocationWrapper function; + private final ExecutorService executor; + + private final FunctionProperties funcProperties; + + private final FunctionCatalog functionCatalog; + + private boolean running; + GrpcServerMessageHandler(FunctionProperties funcProperties, FunctionCatalog functionCatalog) { - this.function = functionCatalog.lookup(funcProperties.getDefinition(), "application/json"); - Assert.notNull(this.function, "Failed to lookup function " + funcProperties.getDefinition()); + this.functionCatalog = functionCatalog; + this.funcProperties = funcProperties; + this.executor = Executors.newCachedThreadPool(); } @Override @SuppressWarnings("unchecked") public void requestReply(GrpcMessage request, StreamObserver responseObserver) { Message message = GrpcUtils.fromGrpcMessage(request); + FunctionInvocationWrapper function = this.resolveFunction(message.getHeaders()); - Message replyMessage = (Message) this.function.apply(message); + Message replyMessage = (Message) function.apply(message); GrpcMessage reply = GrpcUtils.toGrpcMessage(replyMessage); @@ -86,10 +98,12 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { responseObserver.onCompleted(); } + @SuppressWarnings("unchecked") @Override public void serverStream(GrpcMessage request, StreamObserver responseObserver) { Message message = GrpcUtils.fromGrpcMessage(request); - Publisher> replyStream = (Publisher>) this.function.apply(message); + FunctionInvocationWrapper function = this.resolveFunction(message.getHeaders()); + Publisher> replyStream = (Publisher>) function.apply(message); Flux.from(replyStream).doOnNext(replyMessage -> { responseObserver.onNext(GrpcUtils.toGrpcMessage(replyMessage)); }) @@ -104,6 +118,8 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { ServerCallStreamObserver serverCallStreamObserver = (ServerCallStreamObserver) responseObserver; serverCallStreamObserver.disableAutoInboundFlowControl(); + FunctionInvocationWrapper function = this.resolveFunction(null); + AtomicBoolean wasReady = new AtomicBoolean(false); serverCallStreamObserver.setOnReadyHandler(() -> { if (serverCallStreamObserver.isReady() && !wasReady.get()) { @@ -113,26 +129,26 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { } }); - - if (!this.function.isInputTypePublisher()) { + if (!function.isInputTypePublisher()) { throw new UnsupportedOperationException("The client streaming is " + "not supported for functions that accept non-Publisher: " - + this.function); + + function); } - else if (this.function.isOutputTypePublisher()) { + else if (function.isOutputTypePublisher()) { throw new UnsupportedOperationException("The client streaming is " + "not supported for functions that return Publisher: " - + this.function); + + 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); + this.executor.execute(() -> { + Message replyMessage = (Message) function.apply(inputStreamFlux); + if (logger.isDebugEnabled()) { + logger.debug("Function invocation reply: " + replyMessage); + } resultRef.offer(replyMessage); }); @@ -175,6 +191,8 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { ServerCallStreamObserver serverCallStreamObserver = (ServerCallStreamObserver) responseObserver; serverCallStreamObserver.disableAutoInboundFlowControl(); + FunctionInvocationWrapper function = this.resolveFunction(null); + AtomicBoolean wasReady = new AtomicBoolean(false); serverCallStreamObserver.setOnReadyHandler(() -> { if (serverCallStreamObserver.isReady() && !wasReady.get()) { @@ -184,21 +202,21 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { } }); - if (this.function.isInputTypePublisher()) { - if (this.function.isOutputTypePublisher()) { + if (function.isInputTypePublisher()) { + if (function.isOutputTypePublisher()) { return this.biStreamReactive(responseObserver, serverCallStreamObserver); } throw new UnsupportedOperationException("The bi-directional streaming is " + "not supported for functions that accept Publisher but return non-Publisher: " - + this.function); + + function); } else { - if (!this.function.isOutputTypePublisher()) { + if (!function.isOutputTypePublisher()) { return this.biStreamImperative(responseObserver, serverCallStreamObserver, wasReady); } throw new UnsupportedOperationException("The bidirection streaming is " + "not supported for functions that accept non-Publisher but return Publisher: " - + this.function); + + function); } } @@ -212,6 +230,7 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { public void onNext(GrpcMessage request) { try { Message message = GrpcUtils.fromGrpcMessage(request); + FunctionInvocationWrapper function = resolveFunction(message.getHeaders()); Message replyMessage = (Message) function.apply(message); @@ -249,13 +268,38 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { }; } + @Override + public void start() { + this.running = true; + } + + @Override + public void stop() { + this.executor.shutdown(); + try { + Assert.isTrue(this.executor.awaitTermination(5000, TimeUnit.MILLISECONDS), "gRPC Server executor timed out while stopping, " + + "since there are currently executing tasks"); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + this.running = false; + } + + @Override + public boolean isRunning() { + return this.running; + } + @SuppressWarnings("unchecked") private StreamObserver biStreamReactive(StreamObserver responseObserver, ServerCallStreamObserver serverCallStreamObserver) { Many> inputStream = Sinks.many().unicast().onBackpressureBuffer(); Flux> inputStreamFlux = inputStream.asFlux(); - Publisher> outputPublisher = (Publisher>) this.function.apply(inputStreamFlux); + FunctionInvocationWrapper function = this.resolveFunction(null); + + Publisher> outputPublisher = (Publisher>) function.apply(inputStreamFlux); Flux.from(outputPublisher).subscribe(functionResult -> { GrpcMessage outputMessage = GrpcUtils.toGrpcMessage(functionResult); @@ -291,4 +335,16 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase { } }; } + + private FunctionInvocationWrapper resolveFunction(Map headers) { + String functionDefinition = funcProperties.getDefinition(); + if (!CollectionUtils.isEmpty(headers) && headers.containsKey(FunctionProperties.FUNCTION_DEFINITION)) { + functionDefinition = (String) headers.get(FunctionProperties.FUNCTION_DEFINITION); + } + FunctionInvocationWrapper function = this.functionCatalog.lookup(functionDefinition, "application/json"); + Assert.notNull(function, "Failed to lookup function " + funcProperties.getDefinition()); + return function; + } } + + 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 1014708ad..847a2124d 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 @@ -64,6 +64,24 @@ public class GrpcInteractionTests { } } + @Test + public void testRequstReplyFunctionDefinitionInMessage() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + SampleConfiguration.class).web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.function.grpc.port=" + FunctionGrpcProperties.GRPC_PORT)) { + + Message message = MessageBuilder.withPayload("\"hello gRPC\"".getBytes()) + .setHeader("foo", "bar") + .setHeader("spring.cloud.function.definition", "reverse") + .build(); + + Message reply = GrpcUtils.requestReply(message); + + assertThat(reply.getPayload()).isEqualTo("\"CPRg olleh\"".getBytes()); + } + } + @Test public void testBidirectionalStreamWithImperativeFunction() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( @@ -75,13 +93,10 @@ public class GrpcInteractionTests { List> messages = new ArrayList<>(); messages.add(MessageBuilder.withPayload("\"Ricky\"".getBytes()).setHeader("foo", "bar") - .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) .build()); messages.add(MessageBuilder.withPayload("\"Julien\"".getBytes()).setHeader("foo", "bar") - .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) .build()); messages.add(MessageBuilder.withPayload("\"Bubbles\"".getBytes()).setHeader("foo", "bar") - .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) .build()); Flux> clientResponseObserver = @@ -239,6 +254,11 @@ public class GrpcInteractionTests { return v -> v.toUpperCase(); } + @Bean + public Function reverse() { + return v -> new StringBuilder(v).reverse().toString(); + } + @Bean public Function, Flux> uppercaseReactive() { return flux -> flux.map(v -> v.toUpperCase());