0
spring-cloud-function-grpc/.jdk8
Normal file
0
spring-cloud-function-grpc/.jdk8
Normal file
99
spring-cloud-function-grpc/README.md
Normal file
99
spring-cloud-function-grpc/README.md
Normal file
@@ -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
|
||||
```
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-function-rsocket</artifactId>
|
||||
</dependency>
|
||||
```
|
||||
|
||||
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<String, String> 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<Person> message = MessageBuilder.withPayload(p).setHeader("someHeader", "foo").build();
|
||||
|
||||
Message<Employee> result = rsocketRequesterBuilder.tcp("localhost", port)
|
||||
.route("pojoMessageToPojo")
|
||||
.data(message)
|
||||
.retrieveMono(new ParameterizedTypeReference<Message<Employee>>() {})
|
||||
.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
|
||||
92
spring-cloud-function-grpc/pom.xml
Normal file
92
spring-cloud-function-grpc/pom.xml
Normal file
@@ -0,0 +1,92 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>spring-cloud-function-grpc</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<name>Spring Cloud Function gRPC Support</name>
|
||||
<description>Spring Cloud Function gRPC Support</description>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-function-parent</artifactId>
|
||||
<version>3.2.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<properties>
|
||||
<grpc.version>1.16.1</grpc.version>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>io.grpc</groupId>
|
||||
<artifactId>grpc-netty</artifactId>
|
||||
<version>${grpc.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.grpc</groupId>
|
||||
<artifactId>grpc-protobuf</artifactId>
|
||||
<version>${grpc.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.grpc</groupId>
|
||||
<artifactId>grpc-stub</artifactId>
|
||||
<version>${grpc.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-function-context</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<extensions>
|
||||
<extension>
|
||||
<groupId>kr.motd.maven</groupId>
|
||||
<artifactId>os-maven-plugin</artifactId>
|
||||
<version>1.6.1</version>
|
||||
</extension>
|
||||
</extensions>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.xolstice.maven.plugins</groupId>
|
||||
<artifactId>protobuf-maven-plugin</artifactId>
|
||||
<version>0.6.1</version>
|
||||
<configuration>
|
||||
<protocArtifact>
|
||||
com.google.protobuf:protoc:3.3.0:exe:${os.detected.classifier}
|
||||
</protocArtifact>
|
||||
<pluginId>grpc-java</pluginId>
|
||||
<pluginArtifact>
|
||||
io.grpc:protoc-gen-grpc-java:1.4.0:exe:${os.detected.classifier}
|
||||
</pluginArtifact>
|
||||
</configuration>
|
||||
<executions>
|
||||
<execution>
|
||||
<goals>
|
||||
<goal>compile</goal>
|
||||
<goal>compile-custom</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<!-- <plugin> -->
|
||||
<!-- <groupId>org.springframework.boot</groupId> -->
|
||||
<!-- <artifactId>spring-boot-maven-plugin</artifactId> -->
|
||||
<!-- </plugin> -->
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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<GrpcMessage> responseObserver) {
|
||||
Message<byte[]> message = GrpcUtils.fromGrpcMessage(request);
|
||||
Message<byte[]> replyMessage = (Message<byte[]>) 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<GrpcMessage> responseObserver) {
|
||||
//
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public StreamObserver<GrpcMessage> clientStream(
|
||||
// StreamObserver<GrpcMessage> responseObserver) {
|
||||
// return null;
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public StreamObserver<GrpcMessage> biStream(
|
||||
// StreamObserver<GrpcMessage> responseObserver) {
|
||||
// return null;
|
||||
// }
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
@@ -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<String, String> headers) {
|
||||
return GrpcMessage.newBuilder()
|
||||
.setPayload(ByteString.copyFrom(payload))
|
||||
.putAllHeaders(headers)
|
||||
.build();
|
||||
}
|
||||
|
||||
public static GrpcMessage toGrpcMessage(Message<byte[]> message) {
|
||||
Map<String, String> stringHeaders = new HashMap<>();
|
||||
message.getHeaders().forEach((k, v) -> {
|
||||
stringHeaders.put(k, v.toString());
|
||||
});
|
||||
return toGrpcMessage(message.getPayload(), stringHeaders);
|
||||
}
|
||||
|
||||
public static Message<byte[]> fromGrpcMessage(GrpcMessage message) {
|
||||
return MessageBuilder.withPayload(message.getPayload().toByteArray())
|
||||
.copyHeaders(message.getHeadersMap())
|
||||
.build();
|
||||
}
|
||||
|
||||
public static Message<byte[]> requestReply(Message<byte[]> inputMessage) {
|
||||
return requestReply("localhost", FunctionGrpcProperties.GRPC_PORT, inputMessage);
|
||||
}
|
||||
|
||||
public static Message<byte[]> requestReply(String host, int port, Message<byte[]> 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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
syntax = "proto3";
|
||||
option java_multiple_files = true;
|
||||
package org.springframework.cloud.function.grpc;
|
||||
|
||||
message GrpcMessage {
|
||||
bytes payload = 1;
|
||||
map<string, string> 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);
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
|
||||
org.springframework.cloud.function.grpc.GrpcAutoConfiguration
|
||||
@@ -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<byte[]> message = MessageBuilder.withPayload("hello gRPC".getBytes())
|
||||
.setHeader("foo", "bar")
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
|
||||
.build();
|
||||
|
||||
Message<byte[]> reply = GrpcUtils.requestReply(message);
|
||||
|
||||
assertThat(reply.getPayload()).isEqualTo("\"HELLO GRPC\"".getBytes());
|
||||
}
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
public static class SampleConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<String, String> uppercase() {
|
||||
return v -> v.toUpperCase();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
|
||||
Reference in New Issue
Block a user