Croway commented on code in PR #29112:
URL: https://github.com/apache/camel/pull/29112#discussion_r4456929998
##########
components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java:
##########
@@ -30,15 +33,48 @@ public GrpcProducerToRouteControlledStreamObserver(Endpoint
endpoint) {
public void process(Exchange exchange) throws Exception {
StreamObserver<Object> observer = exchange.getProperty(
GrpcConstants.GRPC_RESPONSE_OBSERVER, StreamObserver.class);
+ if (observer == null) {
+ throw new IllegalArgumentException(
+ "The exchange has no " +
GrpcConstants.GRPC_RESPONSE_OBSERVER + " property: it did not come from a"
+ + " gRPC consumer with
routeControlledStreamObserver=true (with"
+ + "
inheritExchangePropertiesForReplies=true for the replies of a gRPC producer)");
+ }
String eventType = exchange.getMessage().getHeader(
GrpcConstants.GRPC_EVENT_TYPE_HEADER, String.class);
+ if (eventType == null) {
+ // nothing is sent: the event to send to the client is unknown
+ throw new IllegalArgumentException(
+ "The " + GrpcConstants.GRPC_EVENT_TYPE_HEADER + " header
must be set to "
+ +
GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT + ", "
+ +
GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED + " or "
+ +
GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR
+ + " to send a reply with
toRouteControlledStreamObserver");
+ }
switch (eventType) {
- case GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT ->
observer.onNext(exchange.getMessage().getBody());
+ case GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT -> onNext(exchange,
observer);
case GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED ->
observer.onCompleted();
case GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR ->
observer.onError((Throwable) exchange.getMessage().getBody());
default -> {
// NO-OP
}
}
}
+
+ /**
+ * Sends the body to the client: each element of a List body, or the body
itself, converted to the response type of
+ * the method that the consumer set as exchange property. When a message
cannot be converted, nothing is sent, the
+ * call ends with an error and the exchange fails, as the client would
otherwise decode the bytes of another message
+ * type (for example the unchanged request) as the response.
+ */
+ private void onNext(Exchange exchange, StreamObserver<Object> observer) {
+ Class<?> responseType =
exchange.getProperty(GrpcConstants.GRPC_RESPONSE_TYPE, Class.class);
+ List<Object> responses;
+ try {
+ responses = GrpcMethodHandler.toResponses(exchange,
exchange.getMessage().getBody(), responseType);
+ } catch (IllegalArgumentException e) {
+
observer.onError(GrpcMethodHandler.toStatusException((GrpcEndpoint)
getEndpoint(), e));
Review Comment:
`muteException` is read from the producer endpoint here, but the option is
`@UriParam(label = "consumer")`. In the endpoint DSL it exists only in
`GrpcEndpointConsumerBuilder`, not in `GrpcEndpointProducerBuilder`.
Meanwhile `muteException=false` on the consumer is ignored in
route-controlled mode. With
`grpc://...?routeControlledStreamObserver=true&muteException=false` and the
default reply producer, the client gets `Status{code=INTERNAL,
description=Exchange processing failed}`. The other consumer path, from #29012,
uses the consumer's setting in `GrpcMethodHandler.toStatusException`.
Could the consumer's setting travel on the exchange next to
`grpcResponseType`? Then the doc could point to the consumer option. Otherwise
the option would need `label = "common"` and regenerated files.
##########
components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java:
##########
@@ -30,15 +33,48 @@ public GrpcProducerToRouteControlledStreamObserver(Endpoint
endpoint) {
public void process(Exchange exchange) throws Exception {
StreamObserver<Object> observer = exchange.getProperty(
GrpcConstants.GRPC_RESPONSE_OBSERVER, StreamObserver.class);
+ if (observer == null) {
+ throw new IllegalArgumentException(
+ "The exchange has no " +
GrpcConstants.GRPC_RESPONSE_OBSERVER + " property: it did not come from a"
+ + " gRPC consumer with
routeControlledStreamObserver=true (with"
+ + "
inheritExchangePropertiesForReplies=true for the replies of a gRPC producer)");
+ }
String eventType = exchange.getMessage().getHeader(
GrpcConstants.GRPC_EVENT_TYPE_HEADER, String.class);
+ if (eventType == null) {
+ // nothing is sent: the event to send to the client is unknown
+ throw new IllegalArgumentException(
+ "The " + GrpcConstants.GRPC_EVENT_TYPE_HEADER + " header
must be set to "
+ +
GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT + ", "
+ +
GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED + " or "
+ +
GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR
+ + " to send a reply with
toRouteControlledStreamObserver");
+ }
switch (eventType) {
- case GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT ->
observer.onNext(exchange.getMessage().getBody());
+ case GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT -> onNext(exchange,
observer);
case GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED ->
observer.onCompleted();
case GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR ->
observer.onError((Throwable) exchange.getMessage().getBody());
default -> {
// NO-OP
}
}
}
+
+ /**
+ * Sends the body to the client: each element of a List body, or the body
itself, converted to the response type of
+ * the method that the consumer set as exchange property. When a message
cannot be converted, nothing is sent, the
+ * call ends with an error and the exchange fails, as the client would
otherwise decode the bytes of another message
+ * type (for example the unchanged request) as the response.
+ */
+ private void onNext(Exchange exchange, StreamObserver<Object> observer) {
+ Class<?> responseType =
exchange.getProperty(GrpcConstants.GRPC_RESPONSE_TYPE, Class.class);
+ List<Object> responses;
+ try {
+ responses = GrpcMethodHandler.toResponses(exchange,
exchange.getMessage().getBody(), responseType);
+ } catch (IllegalArgumentException e) {
+
observer.onError(GrpcMethodHandler.toStatusException((GrpcEndpoint)
getEndpoint(), e));
Review Comment:
After this `onError` the call is closed, but the observer can still be
called. Your `delegationUnchangedRequestIsNotSentAsResponse` test shows it: the
forwarded `onCompleted` exchange fails, and `target/camel-grpc-test.log` has
```
ERROR DefaultErrorHandler - Failed delivery for (...) at
rc-delegation-unchanged ...
java.lang.IllegalStateException: call already closed
WARN GrpcConsumer - Error processing exchange. ... Caused by:
[java.lang.IllegalStateException - call already closed]
```
The test passes because it only checks the client side.
With an error handler that redelivers (`maximumRedeliveries(1)`), the client
gets the right `INTERNAL` error. But the second `observer.onError` here throws
before `throw e`, so the exchange ends with `IllegalStateException: call
already closed` instead of the `IllegalArgumentException` that names both types.
`GrpcRequestPropagationStreamObserver` guards this with its `failed` flag.
Something similar here would avoid both, for example ignoring events once the
observer was closed with `onError`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
