gnodet-bot commented on code in PR #29112:
URL: https://github.com/apache/camel/pull/29112#discussion_r4458096876
##########
components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java:
##########
@@ -30,15 +40,105 @@ 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 eventTypeRequired("must be set to");
+ }
+ // shared by the exchanges of the call; an exchange without it at
least does not send twice on redelivery
+ AtomicBoolean failed =
exchange.getProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER_FAILED,
AtomicBoolean.class);
+ if (failed == null) {
+ failed = new AtomicBoolean();
+ exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER_FAILED,
failed);
+ }
switch (eventType) {
- case GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT ->
observer.onNext(exchange.getMessage().getBody());
- 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
+ case GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT -> onNext(exchange,
observer, failed);
+ case GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED -> {
+ if (!isFailed(exchange, failed, eventType)) {
+ observer.onCompleted();
+ }
}
+ case GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR -> onError(exchange,
observer, failed);
+ default -> throw eventTypeRequired("is " + eventType + ", but must
be");
+ }
+ }
+
+ private static IllegalArgumentException eventTypeRequired(String problem) {
+ return new IllegalArgumentException(
+ "The " + GrpcConstants.GRPC_EVENT_TYPE_HEADER + " header " +
problem + " "
+ +
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");
+ }
+
+ /**
+ * Whether the call already ended with an error, so the event is ignored:
the call is closed and nothing more can be
+ * sent to the client, for example the forwarded onCompleted event after a
reply that did not convert.
+ */
+ private static boolean isFailed(Exchange exchange, AtomicBoolean failed,
String eventType) {
+ if (failed.get()) {
+ LOG.debug("Ignoring the {} event of exchange {}: the call already
ended with an error", eventType,
+ exchange.getExchangeId());
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * 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. A redelivery
fails the same way, without a second error
+ * on the closed call.
+ */
+ private static void onNext(Exchange exchange, StreamObserver<Object>
observer, AtomicBoolean failed) {
+ 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) {
+ if (failed.compareAndSet(false, true)) {
+ // the muteException option of the consumer, muted when unknown
+ boolean muteException =
exchange.getProperty(GrpcConstants.GRPC_MUTE_EXCEPTION, true, Boolean.class);
+
observer.onError(GrpcMethodHandler.toStatusException(muteException, e));
+ }
+ throw e;
+ }
+ if (!isFailed(exchange, failed,
GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT)) {
+ responses.forEach(observer::onNext);
+ }
+ }
+
+ /**
+ * Ends the call with the Throwable in the body. Another body is a mistake
of the route: the call still ends with an
+ * INTERNAL status error with the text of the body as description, so the
client does not wait until its deadline,
+ * and the exchange fails.
+ */
+ private static void onError(Exchange exchange, StreamObserver<Object>
observer, AtomicBoolean failed) {
+ Object body = exchange.getMessage().getBody();
+ if (body instanceof Throwable throwable) {
+ if (!isFailed(exchange, failed,
GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR)) {
+ failed.set(true);
+ observer.onError(throwable);
+ }
+ return;
Review Comment:
⚠️ **Check-then-act race in the `onError` Throwable path.** The `isFailed()`
check and `failed.set(true)` are two separate operations. Under concurrent
processing (`synchronous=false`), a concurrent `onNext` failure doing
`compareAndSet(false, true)` can race between the `isFailed()` check and the
`set(true)` here — both see `failed` as false, both call `observer.onError()`,
and the second call throws `IllegalStateException: call already closed`. The
non-Throwable path below correctly uses `compareAndSet`.
```suggestion
if (body instanceof Throwable throwable) {
if (failed.compareAndSet(false, true)) {
observer.onError(throwable);
}
return;
}
```
--
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]
