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

GH-744 Add initial support for client-side streaming to gRPC

Browse Source
This commit is contained in:
Oleg Zhurakousky
2021-09-24 14:14:35 +02:00
parent c29f4c875d
commit 335d6f0320
4 changed files with 210 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/GrpcServerMessageHandler.java
View File

@@ -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<GrpcMessage> responseObserver) {
//
// }
//
// @Override
// public StreamObserver<GrpcMessage> clientStream(
// StreamObserver<GrpcMessage> responseObserver) {
// return null;
// }
//
@SuppressWarnings("unchecked")
@Override
public StreamObserver<GrpcMessage> clientStream(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("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<Message<byte[]>> inputStream = Sinks.many().unicast().onBackpressureBuffer();
Flux<Message<byte[]>> inputStreamFlux = inputStream.asFlux();
LinkedBlockingQueue<Message<byte[]>> resultRef = new LinkedBlockingQueue<>(1);
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.execute(() -> {
Message<byte[]> replyMessage = (Message<byte[]>) this.function.apply(inputStreamFlux);
System.out.println(replyMessage);
resultRef.offer(replyMessage);
});
return new StreamObserver<GrpcMessage>() {
@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<GrpcMessage> biStream(StreamObserver<GrpcMessage> responseObserver) {
ServerCallStreamObserver<GrpcMessage> serverCallStreamObserver = (ServerCallStreamObserver<GrpcMessage>) 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<GrpcMessage> biStreamReactive(StreamObserver<GrpcMessage> responseObserver,
ServerCallStreamObserver<GrpcMessage> serverCallStreamObserver) {
Many<Message<byte[]>> sink = Sinks.many().unicast().onBackpressureBuffer();
Flux<Message<byte[]>> flux = sink.asFlux();
Many<Message<byte[]>> inputStream = Sinks.many().unicast().onBackpressureBuffer();
Flux<Message<byte[]>> inputStreamFlux = inputStream.asFlux();
Flux<Message<byte[]>> connectedFlux = (Flux<Message<byte[]>>) function.apply(flux);
Publisher<Message<byte[]>> outputPublisher = (Publisher<Message<byte[]>>) 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<GrpcMessage>() {
@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();
}
};

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

@@ -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<byte[]>>} representing input stream.
* @return {@code FluxMessage<byte[]>>} representing output stream
* @return {@code Flux<Message<byte[]>>} representing output stream
*/
public static Flux<Message<byte[]>> biStreaming(Flux<Message<byte[]>> 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<byte[]>>} representing input stream
* @return {@code FluxMessage<byte[]>>} representing output stream
* @return {@code Flux<Message<byte[]>>} representing output stream
*/
public static Flux<Message<byte[]>> biStreaming(String host, int port, Flux<Message<byte[]>> 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<byte[]>>} representing input stream.
* @return {@code Message<byte[]>} representing output
*/
public static Message<byte[]> clientStream(Flux<Message<byte[]>> 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<byte[]>>} representing input stream
* @return {@code Message<byte[]>} representing output
*/
public static Message<byte[]> clientStream(String host, int port, Flux<Message<byte[]>> inputStream) {
ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port)
.usePlaintext().build();
LinkedBlockingQueue<Message<byte[]>> resultRef = new LinkedBlockingQueue<>(1);
StreamObserver<GrpcMessage> responseObserver = new StreamObserver<GrpcMessage>() {
@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<GrpcMessage> 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<GrpcMessage, GrpcMessage> clientResponseObserver(Flux<Message<byte[]>> inputStream, Many<Message<byte[]>> sink) {
return new ClientResponseObserver<GrpcMessage, GrpcMessage>() {
@@ -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);
}

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

@@ -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<Message<byte[]>> 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<byte[]> 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<Flux<String>, 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

2
spring-cloud-function-grpc/src/test/resources/application.properties
View File

@@ -1 +1 @@
logging.level.org.springframework.cloud.function.grpc=DEBUG