davsclaus commented on code in PR #29112:
URL: https://github.com/apache/camel/pull/29112#discussion_r4457607673
##########
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:
Thanks, fixed in 538574afd4fe. The consumer now shares an `AtomicBoolean`
between the exchanges of one call, in the `grpcResponseObserverFailed` exchange
property (one per unary call, one per `DELEGATION` request observer; copied to
reply exchanges like the observer). Once the producer ended the call with
`onError`, it ignores later `onNext`/`onCompleted`/`onError` events (debug
log), so the forwarded `onCompleted` no longer fails. On a redelivery the
conversion runs again and throws the same `IllegalArgumentException`, but no
second `onError` is sent.
`delegationUnchangedRequestIsNotSentAsResponse` now also checks the server
side with an event notifier: the `onNext` exchange fails with the
`IllegalArgumentException` naming both types, the `onCompleted` exchange
completes. New `redeliveryFailsWithTheReplyTypeError`
(`maximumRedeliveries(1)`) checks the client gets `INTERNAL` naming both types
and the exchange ends with the `IllegalArgumentException`. Without the fix both
fail with `IllegalStateException: call already closed`.
_Claude Code on behalf of Claus Ibsen (davsclaus)_
##########
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:
Thanks, done as you suggested in 538574afd4fe. The consumer puts its
`muteException` in the `grpcMuteException` exchange property, next to
`grpcResponseType`, and the producer uses it (muted when the property is
missing). The producer endpoint option is not read any more, and the doc now
points to the consumer option. `GrpcMethodHandler.toStatusException` got an
overload that takes the boolean, so both paths share the code.
The tests now set `muteException=false` on the consumer, not on the reply
endpoint: `unchangedRequestIsNotSentAsResponse`,
`delegationUnchangedRequestIsNotSentAsResponse` and
`redeliveryFailsWithTheReplyTypeError` check the client gets the real
description; they failed with `Exchange processing failed` before.
_Claude Code on behalf of Claus Ibsen (davsclaus)_
--
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]
