SpringCloud/spring-cloud-function
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-744 Add initial support for bi-directional gRPC

Browse Source 
Resolves #744
This commit is contained in:
Oleg Zhurakousky
2021-09-20 14:55:05 +02:00
parent 3b4b7aae5c
commit 9b11199eab
8 changed files with 244 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/FunctionGrpcProperties.java
View File

@@ -31,7 +31,7 @@ public class FunctionGrpcProperties {
/**
* Default gRPC port.
*/
public final static int GRPC_PORT = 55555;
public final static int GRPC_PORT = 6048;
private int port = GRPC_PORT;

2
spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcAutoConfiguration.java
View File

@@ -30,10 +30,10 @@ import org.springframework.context.annotation.Configuration;
*/
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties(FunctionGrpcProperties.class)
@ConditionalOnProperty(name = "spring.cloud.function.grpc.mode", havingValue = "server", matchIfMissing = false)
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);
}

86
spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcMessagingServiceImpl.java
View File

@@ -32,32 +32,48 @@
package org.springframework.cloud.function.grpc;
import java.util.concurrent.atomic.AtomicBoolean;
import io.grpc.Status;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
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;
import org.springframework.util.Assert;
/**
*
* @author Oleg Zhurakousky
* @since 3.2
*
*/
class GrpcMessagingServiceImpl extends MessagingServiceImplBase {
private Log logger = LogFactory.getLog(GrpcMessagingServiceImpl.class);
private final FunctionInvocationWrapper function;
GrpcMessagingServiceImpl(FunctionProperties funcProperties, FunctionCatalog functionCatalog) {
this.function = functionCatalog.lookup(funcProperties.getDefinition(), "application/json");
Assert.notNull(this.function, "Failed to lookup function " + funcProperties.getDefinition());
}
@Override
@SuppressWarnings("unchecked")
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();
}
@@ -74,9 +90,61 @@ class GrpcMessagingServiceImpl extends MessagingServiceImplBase {
// return null;
// }
//
// @Override
// public StreamObserver<GrpcMessage> biStream(
// StreamObserver<GrpcMessage> responseObserver) {
// return null;
// }
@Override
@SuppressWarnings("unchecked")
public StreamObserver<GrpcMessage> biStream(StreamObserver<GrpcMessage> responseObserver) {
ServerCallStreamObserver<GrpcMessage> serverCallStreamObserver = (ServerCallStreamObserver<GrpcMessage>) responseObserver;
serverCallStreamObserver.disableAutoInboundFlowControl();
AtomicBoolean wasReady = new AtomicBoolean(false);
serverCallStreamObserver.setOnReadyHandler(() -> {
if (serverCallStreamObserver.isReady() && !wasReady.get()) {
wasReady.set(true);
logger.info("Server stream is ready");
serverCallStreamObserver.request(1);
}
});
return new StreamObserver<GrpcMessage>() {
@Override
public void onNext(GrpcMessage request) {
try {
Message<byte[]> message = GrpcUtils.fromGrpcMessage(request);
Message<byte[]> replyMessage = (Message<byte[]>) function
.apply(message);
GrpcMessage reply = GrpcUtils.toGrpcMessage(replyMessage);
responseObserver.onNext(reply);
// Check the provided ServerCallStreamObserver to see if it is still
// ready to accept more messages.
if (serverCallStreamObserver.isReady()) {
serverCallStreamObserver.request(1);
}
else {
wasReady.set(false);
}
}
catch (Throwable throwable) {
throwable.printStackTrace();
responseObserver.onError(
Status.UNKNOWN.withDescription("Error handling request").withCause(throwable).asException());
}
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
responseObserver.onCompleted();
}
@Override
public void onCompleted() {
logger.info("Server Stream is complete");
responseObserver.onCompleted();
}
};
}
}

2
spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServer.java
View File

@@ -29,7 +29,7 @@ import org.springframework.context.SmartLifecycle;
class GrpcServer implements SmartLifecycle {
protected Log logger = LogFactory.getLog(GrpcServer.class);
private Log logger = LogFactory.getLog(GrpcServer.class);
private final FunctionGrpcProperties grpcProperties;

105
spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcUtils.java
View File

@@ -22,6 +22,13 @@ import java.util.Map;
import com.google.protobuf.ByteString;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import reactor.core.publisher.Sinks.Many;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
@@ -34,6 +41,8 @@ import org.springframework.messaging.support.MessageBuilder;
*/
public final class GrpcUtils {
private static Log logger = LogFactory.getLog(GrpcUtils.class);
private GrpcUtils() {
}
@@ -64,7 +73,7 @@ public final class GrpcUtils {
}
public static Message<byte[]> requestReply(String host, int port, Message<byte[]> inputMessage) {
ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", port)
ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port)
.usePlaintext().build();
MessagingServiceGrpc.MessagingServiceBlockingStub stub = MessagingServiceGrpc
.newBlockingStub(channel);
@@ -73,4 +82,98 @@ public final class GrpcUtils {
channel.shutdown();
return fromGrpcMessage(response);
}
/**
* Utility method to support bi-directional streaming interaction. Will connect to gRPC server using default host/port,
* otherwise use {@link #biStreaming(String, int, Flux)} method.
*
* Keep in mind that there is no implied relationship between input stream and output stream.
* They are completely independent where one may end before the other.
*
* @param inputStream {@code FluxMessage<byte[]>>} representing input stream.
* @return {@code FluxMessage<byte[]>>} representing output stream
*/
public static Flux<Message<byte[]>> biStreaming(Flux<Message<byte[]>> inputStream) {
return biStreaming("localhost", FunctionGrpcProperties.GRPC_PORT, inputStream);
}
/**
* Utility method to support bi-directional streaming interaction.
* Keep in mind that there is no implied relationship between input stream and output stream.
* They are completely independent where one may end before the other.
*
* @param host gRPC server host name
* @param port gRPC server port
* @param inputStream {@code FluxMessage<byte[]>>} representing input stream
* @return {@code FluxMessage<byte[]>>} representing output stream
*/
public static Flux<Message<byte[]>> biStreaming(String host, int port, Flux<Message<byte[]>> inputStream) {
ManagedChannel channel = ManagedChannelBuilder
.forAddress(host, port)
.usePlaintext().build();
MessagingServiceGrpc.MessagingServiceStub stub = MessagingServiceGrpc
.newStub(channel);
Many<Message<byte[]>> sink = Sinks.many().unicast().onBackpressureBuffer();
ClientResponseObserver<GrpcMessage, GrpcMessage> clientResponseObserver = clientResponseObserver(inputStream, sink);
stub.biStream(clientResponseObserver);
return sink.asFlux().doOnComplete(() -> {
logger.debug("Shutting down channel");
channel.shutdown();
});
}
private static ClientResponseObserver<GrpcMessage, GrpcMessage> clientResponseObserver(Flux<Message<byte[]>> inputStream, Many<Message<byte[]>> sink) {
return new ClientResponseObserver<GrpcMessage, GrpcMessage>() {
ClientCallStreamObserver<GrpcMessage> requestStreamObserver;
@Override
public void beforeStart(ClientCallStreamObserver<GrpcMessage> requestStreamObserver) {
this.requestStreamObserver = requestStreamObserver;
requestStreamObserver.disableAutoInboundFlowControl();
requestStreamObserver.setOnReadyHandler(new Runnable() {
@Override
public void run() {
inputStream
.doOnNext(request -> {
if (logger.isDebugEnabled()) {
logger.debug("Sending message: " + request);
}
requestStreamObserver.onNext(GrpcUtils.toGrpcMessage(request));
})
.doOnComplete(() -> {
requestStreamObserver.onCompleted();
})
.subscribe();
}
});
}
@Override
public void onNext(GrpcMessage message) {
if (logger.isDebugEnabled()) {
logger.debug("Receiving message: " + message);
}
sink.tryEmitNext(fromGrpcMessage(message));
requestStreamObserver.request(1);
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
@Override
public void onCompleted() {
logger.info("Client stream is complete");
sink.tryEmitComplete(); // TODO revisit as this would complete the server stream simply because the client is done.
// Perhaps we need to expose some boolean value when this is desirable
}
};
}
}

49
spring-cloud-function-grpc/src/test/java/org/springframework/cloud/function/grpc/GrpcInteractionTests.java
View File

@@ -16,11 +16,13 @@
package org.springframework.cloud.function.grpc;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -34,18 +36,23 @@ import org.springframework.util.MimeTypeUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
*
* @author Oleg Zhurakousky
*
*/
public class GrpcInteractionTests {
@Test
public void test() {
public void testRequestReply() {
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.port=" + FunctionGrpcProperties.GRPC_PORT,
"--spring.cloud.function.grpc.mode=server")) {
Message<byte[]> message = MessageBuilder.withPayload("hello gRPC".getBytes())
Message<byte[]> message = MessageBuilder.withPayload("\"hello gRPC\"".getBytes())
.setHeader("foo", "bar")
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
.build();
@@ -56,6 +63,38 @@ public class GrpcInteractionTests {
}
}
@Test
public void testBidirectionalStream() {
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="
+ FunctionGrpcProperties.GRPC_PORT,
"--spring.cloud.function.grpc.mode=server")) {
List<Message<byte[]>> 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<Message<byte[]>> clientResponseObserver =
GrpcUtils.biStreaming("localhost", FunctionGrpcProperties.GRPC_PORT, Flux.fromIterable(messages));
List<Message<byte[]>> results = clientResponseObserver.collectList().block(Duration.ofSeconds(1));
assertThat(results.size()).isEqualTo(3);
assertThat(results.get(0).getPayload()).isEqualTo("\"RICKY\"".getBytes());
assertThat(results.get(1).getPayload()).isEqualTo("\"JULIEN\"".getBytes());
assertThat(results.get(2).getPayload()).isEqualTo("\"BUBBLES\"".getBytes());
}
}
@EnableAutoConfiguration
public static class SampleConfiguration {