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

GH-744 Make default wait time indefinite for client-side streaming when awaiting response

Browse Source
This commit is contained in:
Oleg Zhurakousky
2021-09-24 16:02:30 +02:00
parent f01f23fb64
commit bc3a4ea7ca
2 changed files with 4 additions and 4 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/GrpcServerMessageHandler.java
View File

@@ -153,7 +153,7 @@ class GrpcServerMessageHandler extends MessagingServiceImplBase {
logger.info("gRPC Server has finished receiving data."); logger.info("gRPC Server has finished receiving data.");
inputStream.tryEmitComplete(); inputStream.tryEmitComplete();
try { try {
responseObserver.onNext(GrpcUtils.toGrpcMessage(resultRef.poll(5000, TimeUnit.MILLISECONDS))); responseObserver.onNext(GrpcUtils.toGrpcMessage(resultRef.poll(Integer.MAX_VALUE, TimeUnit.MILLISECONDS)));
} }
catch (InterruptedException e) { catch (InterruptedException e) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();

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

@@ -191,7 +191,7 @@ final class GrpcUtils {
}).subscribe(); }).subscribe();
try { try {
return resultRef.poll(5000, TimeUnit.MILLISECONDS); return resultRef.poll(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
} }
catch (InterruptedException ie) { catch (InterruptedException ie) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
@@ -215,7 +215,7 @@ final class GrpcUtils {
inputStream inputStream
.doOnNext(request -> { .doOnNext(request -> {
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("Sending message: " + request); logger.debug("Streaming message to function: " + request);
} }
requestStreamObserver.onNext(GrpcUtils.toGrpcMessage(request)); requestStreamObserver.onNext(GrpcUtils.toGrpcMessage(request));
}) })
@@ -230,7 +230,7 @@ final class GrpcUtils {
@Override @Override
public void onNext(GrpcMessage message) { public void onNext(GrpcMessage message) {
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("Receiving message: " + message); logger.debug("Streaming message from function: " + message);
} }
sink.tryEmitNext(fromGrpcMessage(message)); sink.tryEmitNext(fromGrpcMessage(message));
requestStreamObserver.request(1); requestStreamObserver.request(1);