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

gRPC cleanup

Browse Source
This commit is contained in:
Oleg Zhurakousky
2021-10-19 09:10:14 +02:00
parent c8c0357786
commit 6a48a10d94
2 changed files with 1 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,9 +16,6 @@
package org.springframework.cloud.function.grpc; package org.springframework.cloud.function.grpc;
import java.io.Closeable;
import java.io.IOException;
import java.util.HashMap; import java.util.HashMap;
import java.util.Iterator; import java.util.Iterator;
import java.util.Map; import java.util.Map;
@@ -241,7 +238,6 @@ final class GrpcUtils {
e.printStackTrace(); e.printStackTrace();
responseObserver.onError(Status.UNKNOWN.withDescription("Error handling request") responseObserver.onError(Status.UNKNOWN.withDescription("Error handling request")
.withCause(e).asRuntimeException()); .withCause(e).asRuntimeException());
// channel.shutdownNow();
}) })
.subscribe(); .subscribe();
@@ -252,9 +248,6 @@ final class GrpcUtils {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
throw new IllegalStateException(ie); throw new IllegalStateException(ie);
} }
// finally {
// channel.shutdownNow();
// }
} }
private static ClientResponseObserver<GrpcSpringMessage, GrpcSpringMessage> clientResponseObserver(Flux<Message<byte[]>> inputStream, Many<Message<byte[]>> sink) { private static ClientResponseObserver<GrpcSpringMessage, GrpcSpringMessage> clientResponseObserver(Flux<Message<byte[]>> inputStream, Many<Message<byte[]>> sink) {

8
spring-cloud-function-adapters/spring-cloud-function-grpc/src/main/java/org/springframework/cloud/function/grpc/MessageHandlingHelper.java
View File

@@ -47,7 +47,7 @@ import org.springframework.util.CollectionUtils;
/** /**
* *
* @author Oleg Zhurakousky * @author Oleg Zhurakousky
* * @since 3.2
*/ */
public class MessageHandlingHelper<T extends GeneratedMessageV3> implements SmartLifecycle { public class MessageHandlingHelper<T extends GeneratedMessageV3> implements SmartLifecycle {
@@ -147,7 +147,6 @@ public class MessageHandlingHelper<T extends GeneratedMessageV3> implements Smar
@Override @Override
public void onError(Throwable t) { public void onError(Throwable t) {
t.printStackTrace(); t.printStackTrace();
// responseObserver.onCompleted();
responseObserver.onError(Status.UNKNOWN.withDescription("Error handling request") responseObserver.onError(Status.UNKNOWN.withDescription("Error handling request")
.withCause(t).asRuntimeException()); .withCause(t).asRuntimeException());
} }
@@ -192,8 +191,6 @@ public class MessageHandlingHelper<T extends GeneratedMessageV3> implements Smar
UnsupportedOperationException ex = new UnsupportedOperationException("The bi-directional streaming is " UnsupportedOperationException ex = new UnsupportedOperationException("The bi-directional streaming is "
+ "not supported for functions that accept Publisher but return non-Publisher: " + "not supported for functions that accept Publisher but return non-Publisher: "
+ function); + function);
// responseObserver.onError(Status.UNKNOWN.withDescription("Error handling request")
// .withCause(ex).asException());
responseObserver.onCompleted(); responseObserver.onCompleted();
throw ex; throw ex;
} }
@@ -205,8 +202,6 @@ public class MessageHandlingHelper<T extends GeneratedMessageV3> implements Smar
UnsupportedOperationException ex = new UnsupportedOperationException("The bidirection streaming is " UnsupportedOperationException ex = new UnsupportedOperationException("The bidirection streaming is "
+ "not supported for functions that accept non-Publisher but return Publisher: " + "not supported for functions that accept non-Publisher but return Publisher: "
+ function); + function);
// responseObserver.onError(Status.UNKNOWN.withDescription("Error handling request")
// .withCause(ex).asException());
responseObserver.onCompleted(); responseObserver.onCompleted();
throw ex; throw ex;
} }
@@ -243,7 +238,6 @@ public class MessageHandlingHelper<T extends GeneratedMessageV3> implements Smar
@Override @Override
public void onError(Throwable t) { public void onError(Throwable t) {
t.printStackTrace(); t.printStackTrace();
//responseObserver.onCompleted();
inputStream.tryEmitComplete(); inputStream.tryEmitComplete();
responseObserver.onError(Status.UNKNOWN.withDescription("Error handling request") responseObserver.onError(Status.UNKNOWN.withDescription("Error handling request")
.withCause(t).asException()); .withCause(t).asException());