diff --git a/spring-cloud-function-grpc/.jdk8 b/spring-cloud-function-grpc/.jdk8 new file mode 100644 index 000000000..e69de29bb diff --git a/spring-cloud-function-grpc/README.md b/spring-cloud-function-grpc/README.md new file mode 100644 index 000000000..5ce6c0c2e --- /dev/null +++ b/spring-cloud-function-grpc/README.md @@ -0,0 +1,99 @@ +### Introduction + +Spring Cloud Function allows you to invoke function via [RSocket](https://rsocket.io/). While you can read more about RSocket and it’s java +implementation [here](https://github.com/rsocket/rsocket-java), this section will describe the parts relevant to Spring Cloud Function integration. + +### Programming model +From the user perspective bringing RSocket does not change the implementation of functions or any of its features, such as type conversion, +composition, POJO functions etc. +And while RSocket allows first class reactive interaction over the network supporting important reactive features such as back pressure, +users of Spring Cloud Function still have freedom to implement their business logic using reactive or imperative functions delegating any +adjustment needed to apply proper invocation model to the framework. + +To use RSocket integration all you need is to add `spring-cloud-function-rsocket` dependency to your classpath +``` + + org.springframework.cloud + spring-cloud-function-rsocket + +``` + +To interact with functions via RSocket we rely on Spring Boot support for RSocket and `RSocketRequester.Builder` API. +The code below shows the key parts and you can get more details on various interaction models +from [this test case](https://github.com/spring-cloud/spring-cloud-function/blob/master/spring-cloud-function-rsocket/src/test/java/org/springframework/cloud/function/rsocket/RSocketAutoConfigurationTests.java). + + +``` +@Bean +public Function uppercase() { + return v -> v.toUpperCase(); +} + +. . . + +RSocketRequester.Builder rsocketRequesterBuilder = + applicationContext.getBean(RSocketRequester.Builder.class); + +rsocketRequesterBuilder.tcp("localhost", port) + .route(“uppercase") + .data("\"hello\"") + .retrieveMono(String.class) + .subscribe(System.out::println); +``` + +Once connected to RSocket we use `route` operation to specify which function we want to invoke providing the actual +payload via `data` operation. Then we use one of the `retrieve` operations that best suits our desired interaction +(RSocket supports multiple interaction models such as fire-and-forget, request-reply etc.) + +#### Order of priority for routing instructions + +As you can see from the preceding examples, we provide function definition as a value to `route(..)` operator of `RSocketRequester.Builder`. +However that is not the only way. You can also use standard `spring.cloud.function.definition` property as well as `spring.cloud.function.routing-expression` or property or `MessageRoutingCallback` on the server side of the RSocket interaction (see "Function Routing and Filtering" section of reference manual). +This raises a question of _order_ and _priorities_ when it comes to reconsiling a conflict in the event several ways of providing definition are used. So it is a mater of clearly stating the rule whcih is: + +***1 - MessageRoutingCallback*** +The `MessageRoutingCallback` takes precedence over all other ways of providing function definition resolution. + +***2 - spring.cloud.function.routing-expression*** +The `spring.cloud.function.routing-expression` property takes next precedence. So, in the event you may have also use `route(..)` operator or `spring.cloud.function.definition` property, they will be ignored if `spring.cloud.function.routing-expression` property is provided. + +***3 - route(..)*** +The next in line is `route(..)` operator. So in the event there are no `spring.cloud.function.routing-expression` property but you defined `spring.cloud.function.definition` property, it will be ignored in favor of definition provided by the `route(..)` operator. + +***4 - spring.cloud.function.definition*** +The `spring.cloud.function.definition` property is the last in the list allowing you to simply `route("")` to empty string. + + +### Messaging + +If you want to provide and/or receive additional information that you would normally communicate via Message headers you can send and receive Spring `Message`. +For example, the following tests case demonstrates how you can accomplish that. +``` +Person p = new Person(); +p.setName("Ricky"); +Message message = MessageBuilder.withPayload(p).setHeader("someHeader", "foo").build(); + +Message result = rsocketRequesterBuilder.tcp("localhost", port) + .route("pojoMessageToPojo") + .data(message) + .retrieveMono(new ParameterizedTypeReference>() {}) + .block(); +``` +Aside from sending `Message`, note the usage of `ParameterizedTypeReference` to specify that we want not only `Message` in return but also `Message` with specific payload type. + +### Function Composition over RSocket (Distributed Function Composition) + +By now you shoudl be familiar with the standard function composition feature (e.g., `functionA|functionB|functionC`). This feature allows you to compose several co-located functions into one. But what if these functions are not co-located and instead separated by the network? + +With RSocket and our _distributed function composition_ feature you can still do it. So let's look at the example. + +Let's say we have `uppercase` function available to you locally and `reverse` function exposed via separate RSocket and you wan to compose `uppercase` and `reverse` into a single function. Had they been both available locally it would have been as simple as `uppercase|reverse`, but given that `reverse` function is not locally available we need a way to specify that in our composition instruction. For that we're using _redirect_ operator. So it woudl look like this `uppercase>localhost:2222`, where `localhost:2222` is the host/port combination where `reverse` function is hosted. +What's interesting is that remote function can in itself be a result of function composition (local or remote), so effectively you are composing `uppercase` with whatever function definition (which could be composition) that is running on `localhost:2222`. + +The complete example is available in [this test case](https://github.com/spring-cloud/spring-cloud-function/blob/0e3a27a392f5c69727d909db26c2ba6aa0344cfd/spring-cloud-function-rsocket/src/test/java/org/springframework/cloud/function/rsocket/RSocketAutoConfigurationTests.java#L371). +And as you can see it is a bit more complex to showcase this feature. In this test we are composing `reverse` function with `uppercase|concat` function running remotely and then with `wrap` function running locally as if `reverse|uppercase|concat|wrap`. +So you can see `--spring.cloud.function.definition=reverse>localhost:" + portA + "|wrap"` where `localhost:" + portA` points to another application context instance with `--spring.cloud.function.definition=uppercase|concat`. The result of the `reverse` function are sent to `uppercase|concat` function via RSocket and the result of that are fed into `wrap` function. + +### Samples + +You can also look at one of the [RSocket samples](https://github.com/spring-cloud/spring-cloud-function/tree/master/spring-cloud-function-samples/function-sample-cloudevent-rsocket) that is also introduces you to Cloud Events \ No newline at end of file diff --git a/spring-cloud-function-grpc/pom.xml b/spring-cloud-function-grpc/pom.xml new file mode 100644 index 000000000..85affe799 --- /dev/null +++ b/spring-cloud-function-grpc/pom.xml @@ -0,0 +1,92 @@ + + + 4.0.0 + + spring-cloud-function-grpc + jar + Spring Cloud Function gRPC Support + Spring Cloud Function gRPC Support + + + org.springframework.cloud + spring-cloud-function-parent + 3.2.0-SNAPSHOT + + + + 1.16.1 + + + + + io.grpc + grpc-netty + ${grpc.version} + + + io.grpc + grpc-protobuf + ${grpc.version} + + + io.grpc + grpc-stub + ${grpc.version} + + + org.springframework.cloud + spring-cloud-function-context + + + org.springframework.boot + spring-boot-configuration-processor + true + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + kr.motd.maven + os-maven-plugin + 1.6.1 + + + + + org.xolstice.maven.plugins + protobuf-maven-plugin + 0.6.1 + + + com.google.protobuf:protoc:3.3.0:exe:${os.detected.classifier} + + grpc-java + + io.grpc:protoc-gen-grpc-java:1.4.0:exe:${os.detected.classifier} + + + + + + compile + compile-custom + + + + + + + + + + + + diff --git a/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/FunctionGrpcProperties.java b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/FunctionGrpcProperties.java new file mode 100644 index 000000000..c59c41398 --- /dev/null +++ b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/FunctionGrpcProperties.java @@ -0,0 +1,48 @@ +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.function.grpc; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.function.context.FunctionProperties; + +/** + * + * @author Oleg Zhurakousky + * @since 3.2 + * + */ +@ConfigurationProperties(prefix = FunctionProperties.PREFIX + ".grpc") +public class FunctionGrpcProperties { + + /** + * Default gRPC port. + */ + public final static int GRPC_PORT = 55555; + + private int port = GRPC_PORT; + + /** + * Grpc Server port. + */ + public int getPort() { + return this.port; + } + + public void setPort(int port) { + this.port = port; + } +} diff --git a/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcAutoConfiguration.java b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcAutoConfiguration.java new file mode 100644 index 000000000..a1462c65e --- /dev/null +++ b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcAutoConfiguration.java @@ -0,0 +1,46 @@ +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.function.grpc; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.FunctionProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * + * @author Oleg Zhurakousky + * @since 3.2 + */ +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(FunctionGrpcProperties.class) +public class GrpcAutoConfiguration { + + @Bean + @ConditionalOnProperty(name = "spring.cloud.function.grpc.mode", havingValue = "server", matchIfMissing = false) + public GrpcServer grpcServer(FunctionGrpcProperties grpcProperties, GrpcMessagingServiceImpl grpcMessagingService) { + return new GrpcServer(grpcProperties, grpcMessagingService); + } + + + @Bean + public GrpcMessagingServiceImpl grpcMessageService(FunctionProperties funcProperties, FunctionCatalog functionCatalog) { + return new GrpcMessagingServiceImpl(funcProperties, functionCatalog); + } +} diff --git a/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcMessagingServiceImpl.java b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcMessagingServiceImpl.java new file mode 100644 index 000000000..3187b8561 --- /dev/null +++ b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcMessagingServiceImpl.java @@ -0,0 +1,82 @@ +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.function.grpc; + +import io.grpc.stub.StreamObserver; + +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.messaging.Message; + +class GrpcMessagingServiceImpl extends MessagingServiceImplBase { + + private final FunctionInvocationWrapper function; + + GrpcMessagingServiceImpl(FunctionProperties funcProperties, FunctionCatalog functionCatalog) { + this.function = functionCatalog.lookup(funcProperties.getDefinition(), "application/json"); + } + + + @Override + public void requestReply(GrpcMessage request, StreamObserver responseObserver) { + Message message = GrpcUtils.fromGrpcMessage(request); + Message replyMessage = (Message) this.function.apply(message); + + GrpcMessage reply = GrpcUtils.toGrpcMessage(replyMessage); + /* + * The above is effectively echo. This is where we plug in function invocation + */ + responseObserver.onNext(reply); + responseObserver.onCompleted(); + } +// +// @Override +// public void serverStream(GrpcMessage request, +// StreamObserver responseObserver) { +// +// } +// +// @Override +// public StreamObserver clientStream( +// StreamObserver responseObserver) { +// return null; +// } +// +// @Override +// public StreamObserver biStream( +// StreamObserver responseObserver) { +// return null; +// } +} diff --git a/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServer.java b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServer.java new file mode 100644 index 000000000..93a53c857 --- /dev/null +++ b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServer.java @@ -0,0 +1,76 @@ +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.function.grpc; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import io.grpc.Server; +import io.grpc.ServerBuilder; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.cloud.function.grpc.MessagingServiceGrpc.MessagingServiceImplBase; +import org.springframework.context.SmartLifecycle; + +class GrpcServer implements SmartLifecycle { + + protected Log logger = LogFactory.getLog(GrpcServer.class); + + private final FunctionGrpcProperties grpcProperties; + + private final MessagingServiceImplBase grpcMessageService; + + private final ExecutorService executor = Executors.newSingleThreadExecutor(); + + private Server server; + + GrpcServer(FunctionGrpcProperties grpcProperties, MessagingServiceImplBase grpcMessageService) { + this.grpcProperties = grpcProperties; + this.grpcMessageService = grpcMessageService; + } + + @Override + public void start() { + this.executor.execute(() -> { + try { + this.server = ServerBuilder.forPort(this.grpcProperties.getPort()) + .addService(this.grpcMessageService) + .build(); + + logger.info("Starting gRPC server"); + this.server.start(); + logger.info("gRPC server is listening on port " + this.grpcProperties.getPort()); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + }); + } + + @Override + public void stop() { + logger.info("Shutting down gRPC server"); + this.server.shutdown(); + this.executor.shutdown(); + } + + @Override + public boolean isRunning() { + return this.server != null && !this.server.isShutdown(); + } +} 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 new file mode 100644 index 000000000..67906cc0b --- /dev/null +++ b/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcUtils.java @@ -0,0 +1,76 @@ +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.function.grpc; + +import java.util.HashMap; +import java.util.Map; + +import com.google.protobuf.ByteString; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; + +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; + +/** + * + * @author Oleg Zhurakousky + * @since 3.2 + * + */ +public final class GrpcUtils { + + private GrpcUtils() { + + } + + public static GrpcMessage toGrpcMessage(byte[] payload, Map headers) { + return GrpcMessage.newBuilder() + .setPayload(ByteString.copyFrom(payload)) + .putAllHeaders(headers) + .build(); + } + + public static GrpcMessage toGrpcMessage(Message message) { + Map stringHeaders = new HashMap<>(); + message.getHeaders().forEach((k, v) -> { + stringHeaders.put(k, v.toString()); + }); + return toGrpcMessage(message.getPayload(), stringHeaders); + } + + public static Message fromGrpcMessage(GrpcMessage message) { + return MessageBuilder.withPayload(message.getPayload().toByteArray()) + .copyHeaders(message.getHeadersMap()) + .build(); + } + + public static Message requestReply(Message inputMessage) { + return requestReply("localhost", FunctionGrpcProperties.GRPC_PORT, inputMessage); + } + + public static Message requestReply(String host, int port, Message inputMessage) { + ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 55555) + .usePlaintext().build(); + MessagingServiceGrpc.MessagingServiceBlockingStub stub = MessagingServiceGrpc + .newBlockingStub(channel); + + GrpcMessage response = stub.requestReply(toGrpcMessage(inputMessage)); + channel.shutdown(); + return fromGrpcMessage(response); + } +} diff --git a/spring-cloud-function-grpc/src/main/proto/MessageService.proto b/spring-cloud-function-grpc/src/main/proto/MessageService.proto new file mode 100644 index 000000000..7da6fbcad --- /dev/null +++ b/spring-cloud-function-grpc/src/main/proto/MessageService.proto @@ -0,0 +1,18 @@ +syntax = "proto3"; +option java_multiple_files = true; +package org.springframework.cloud.function.grpc; + +message GrpcMessage { + bytes payload = 1; + map headers = 2; +} + +service MessagingService { + rpc biStream(stream GrpcMessage) returns (stream GrpcMessage); + + rpc clientStream(stream GrpcMessage) returns (GrpcMessage); + + rpc serverStream(GrpcMessage) returns (stream GrpcMessage); + + rpc requestReply(GrpcMessage) returns (GrpcMessage); +} \ No newline at end of file diff --git a/spring-cloud-function-grpc/src/main/resources/META-INF/spring.factories b/spring-cloud-function-grpc/src/main/resources/META-INF/spring.factories new file mode 100644 index 000000000..3798e8579 --- /dev/null +++ b/spring-cloud-function-grpc/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ +org.springframework.cloud.function.grpc.GrpcAutoConfiguration 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 new file mode 100644 index 000000000..aa03b5a45 --- /dev/null +++ b/spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java @@ -0,0 +1,67 @@ +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.function.grpc; + + + +import java.util.function.Function; + +import org.junit.jupiter.api.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.MimeTypeUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +public class GrpcInteractionTests { + + @Test + public void test() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + SampleConfiguration.class).web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.function.definition=uppercase", + "--spring.cloud.function.grpc.port=55555", + "--spring.cloud.function.grpc.mode=server")) { + + Message message = MessageBuilder.withPayload("hello gRPC".getBytes()) + .setHeader("foo", "bar") + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) + .build(); + + Message reply = GrpcUtils.requestReply(message); + + assertThat(reply.getPayload()).isEqualTo("\"HELLO GRPC\"".getBytes()); + } + } + + @EnableAutoConfiguration + public static class SampleConfiguration { + + @Bean + public Function uppercase() { + return v -> v.toUpperCase(); + } + } +} diff --git a/spring-cloud-function-grpc/src/test/resources/application.properties b/spring-cloud-function-grpc/src/test/resources/application.properties new file mode 100644 index 000000000..8b1378917 --- /dev/null +++ b/spring-cloud-function-grpc/src/test/resources/application.properties @@ -0,0 +1 @@ +