[camel] 01/02: CAMEL-14893: Handling onError Thanks, son, kyoungduk for implementation

[dmvolod]

This is an automated email from the ASF dual-hosted git repository.

dmvolod pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git

commit fe9f41f4848c9931bc55a611e0265617530fd968
Author: Dmitry Volodin <dmvo...@gmail.com>
AuthorDate: Tue Apr 14 19:18:55 2020 +0300

    CAMEL-14893: Handling onError
    Thanks, son, kyoungduk for implementation
---
 .../component/grpc/server/GrpcMethodHandler.java   | 24 ++++++++++++++++------
 1 file changed, 18 insertions(+), 6 deletions(-)

diff --git 
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcMethodHandler.java
 
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcMethodHandler.java
index 14c2eda..6364e65 100644
--- 
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcMethodHandler.java
+++ 
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcMethodHandler.java
@@ -21,6 +21,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import javassist.util.proxy.MethodHandler;
 import org.apache.camel.Exchange;
@@ -64,14 +65,25 @@ public class GrpcMethodHandler implements MethodHandler {
             }
             
             StreamObserver<Object> responseObserver = 
(StreamObserver<Object>)args[1];
-            Object responseBody = exchange.getIn().getBody();
-            if (responseBody instanceof List) {
-                List<Object> responseList = (List<Object>)responseBody;
-                responseList.forEach(responseObserver::onNext);
+            
+            if (exchange.isFailed()) {
+                responseObserver.onError(Status.INTERNAL
+                        .withDescription(exchange.getException().getMessage())
+                        // This can be attached to the Status locally, but NOT 
transmitted to the client!
+                        .withCause(exchange.getException())
+                        .asRuntimeException());
             } else {
-                responseObserver.onNext(responseBody);
+                Object responseBody = exchange.getIn().getBody();
+                if (responseBody instanceof List) {
+                    List<Object> responseList = (List<Object>) responseBody;
+                    responseList.forEach(responseItem -> {
+                        responseObserver.onNext(responseItem);
+                    });
+                } else {
+                    responseObserver.onNext(responseBody);
+                }
+                responseObserver.onCompleted();
             }
-            responseObserver.onCompleted();
         } else if (args.length == 1 && args[0] instanceof StreamObserver) {
             // Single incoming parameter is instance of the 
io.grpc.stub.StreamObserver
             final StreamObserver<Object> responseObserver = 
(StreamObserver<Object>)args[0];