This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 65b4e7c11494 CAMEL-26515: camel-grpc - the
toRouteControlledStreamObserver producer converts the reply to the response
type, fails clearly on a bad event or observer, and honours the consumer
muteException (#29112)
65b4e7c11494 is described below
commit 65b4e7c1149403ccbc984fb58adadebbb72f7e33
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Nov 4 20:00:19 2026 +0100
CAMEL-26515: camel-grpc - the toRouteControlledStreamObserver producer
converts the reply to the response type, fails clearly on a bad event or
observer, and honours the consumer muteException (#29112)
- CAMEL-26515: camel-grpc - the toRouteControlledStreamObserver producer
converts the reply to the response type of the method, or fails the call
- CAMEL-26515: camel-grpc - the toRouteControlledStreamObserver producer
fails with a clear error without the event type header or the response observer
- CAMEL-26515: camel-grpc - the toRouteControlledStreamObserver producer
ignores events after it failed the call, uses the consumer's muteException and
rejects unknown events and non-Throwable errors
- CAMEL-26515: camel-grpc - end the call once when an onError event races a
reply that does not convert
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../apache/camel/catalog/docs/grpc-component.adoc | 25 +-
.../camel-grpc/src/main/docs/grpc-component.adoc | 25 +-
.../apache/camel/component/grpc/GrpcConstants.java | 16 +
...rpcProducerToRouteControlledStreamObserver.java | 113 ++++++-
.../component/grpc/server/GrpcMethodHandler.java | 77 ++++-
.../GrpcRequestDelegationStreamObserver.java | 16 +-
.../grpc/GrpcRouteControlledReplyTypeTest.java | 375 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
8 files changed, 630 insertions(+), 25 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/grpc-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/grpc-component.adoc
index 9cc188ba41a3..918d1db67603 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/grpc-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/grpc-component.adoc
@@ -80,8 +80,29 @@ method, or a `List` of them for a streaming response. A
message without a body s
A body of another type is converted to the response message type with a Camel
type converter. When there is no such
conversion, nothing is sent, and the call ends with an `INTERNAL` status
error. For example, a route that leaves the
-request message unchanged fails the call instead of sending the request as the
response. This does not apply with
-`routeControlledStreamObserver=true`, where the route sends the responses
itself.
+request message unchanged fails the call instead of sending the request as the
response.
+
+With `routeControlledStreamObserver=true` the route sends the responses
itself, with a producer endpoint with
+`toRouteControlledStreamObserver=true`. That producer applies the same
conversion to the body of each `onNext` event
+(each element of a `List` body is sent as a message). When a message cannot be
converted, nothing is sent, the call
+ends with an `INTERNAL` status error and the exchange fails. A redelivery of
the exchange fails the same way. The error
+description is muted unless the consumer has `muteException=false`. The
consumer passes the response message type in
+the `grpcResponseType` exchange property and its `muteException` option in the
`grpcMuteException` exchange property;
+when `grpcResponseType` is not on the exchange, the body is sent as is. Once
the producer ended the call with an
+error, it ignores the later events of the call, such as a forwarded
`onCompleted`, as nothing more can be sent to the
+client. Calls on the `StreamObserver` from the `grpcResponseObserver` exchange
property in the route's own code are not
+checked.
+
+The body of an `onError` event must be a `Throwable`, which ends the call. Any
other body, or no body, fails the
+exchange with an `IllegalArgumentException`; the call still ends with an
`INTERNAL` status error that has the text of
+the body as description, so the client does not wait until its deadline.
+
+The producer sends the event named by the `CamelGrpcEventType` header
(`onNext`, `onCompleted` or `onError`). The
+consumer sets this header for the `DELEGATION` events, and the reply exchanges
of a `streamRepliesTo` producer have it.
+When a route calls the producer with an exchange without the header, for
example in a unary call, it must set the header
+itself: without it, or with another value, nothing is sent and the exchange
fails with an `IllegalArgumentException`.
+An exchange without the `grpcResponseObserver` exchange property, that did not
come from a consumer with
+`routeControlledStreamObserver=true`, also fails with an
`IllegalArgumentException`.
=== gRPC producer resource type mapping
diff --git a/components/camel-grpc/src/main/docs/grpc-component.adoc
b/components/camel-grpc/src/main/docs/grpc-component.adoc
index 9cc188ba41a3..918d1db67603 100644
--- a/components/camel-grpc/src/main/docs/grpc-component.adoc
+++ b/components/camel-grpc/src/main/docs/grpc-component.adoc
@@ -80,8 +80,29 @@ method, or a `List` of them for a streaming response. A
message without a body s
A body of another type is converted to the response message type with a Camel
type converter. When there is no such
conversion, nothing is sent, and the call ends with an `INTERNAL` status
error. For example, a route that leaves the
-request message unchanged fails the call instead of sending the request as the
response. This does not apply with
-`routeControlledStreamObserver=true`, where the route sends the responses
itself.
+request message unchanged fails the call instead of sending the request as the
response.
+
+With `routeControlledStreamObserver=true` the route sends the responses
itself, with a producer endpoint with
+`toRouteControlledStreamObserver=true`. That producer applies the same
conversion to the body of each `onNext` event
+(each element of a `List` body is sent as a message). When a message cannot be
converted, nothing is sent, the call
+ends with an `INTERNAL` status error and the exchange fails. A redelivery of
the exchange fails the same way. The error
+description is muted unless the consumer has `muteException=false`. The
consumer passes the response message type in
+the `grpcResponseType` exchange property and its `muteException` option in the
`grpcMuteException` exchange property;
+when `grpcResponseType` is not on the exchange, the body is sent as is. Once
the producer ended the call with an
+error, it ignores the later events of the call, such as a forwarded
`onCompleted`, as nothing more can be sent to the
+client. Calls on the `StreamObserver` from the `grpcResponseObserver` exchange
property in the route's own code are not
+checked.
+
+The body of an `onError` event must be a `Throwable`, which ends the call. Any
other body, or no body, fails the
+exchange with an `IllegalArgumentException`; the call still ends with an
`INTERNAL` status error that has the text of
+the body as description, so the client does not wait until its deadline.
+
+The producer sends the event named by the `CamelGrpcEventType` header
(`onNext`, `onCompleted` or `onError`). The
+consumer sets this header for the `DELEGATION` events, and the reply exchanges
of a `streamRepliesTo` producer have it.
+When a route calls the producer with an exchange without the header, for
example in a unary call, it must set the header
+itself: without it, or with another value, nothing is sent and the exchange
fails with an `IllegalArgumentException`.
+An exchange without the `grpcResponseObserver` exchange property, that did not
come from a consumer with
+`routeControlledStreamObserver=true`, also fails with an
`IllegalArgumentException`.
=== gRPC producer resource type mapping
diff --git
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConstants.java
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConstants.java
index bbb140e20d81..ef1a8b12a557 100644
---
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConstants.java
+++
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConstants.java
@@ -70,4 +70,20 @@ public interface GrpcConstants {
*/
String GRPC_BINDABLE_SERVICE_FACTORY_NAME = "grpcBindableServiceFactory";
String GRPC_RESPONSE_OBSERVER = "grpcResponseObserver";
+ /*
+ * The exchange property with the response message type of the called
method, set by a consumer with
+ * routeControlledStreamObserver=true: the toRouteControlledStreamObserver
producer converts each reply to it
+ */
+ String GRPC_RESPONSE_TYPE = "grpcResponseType";
+ /*
+ * The exchange property with the muteException option of a consumer with
routeControlledStreamObserver=true: the
+ * toRouteControlledStreamObserver producer uses it for the description of
the errors it sends to the client
+ */
+ String GRPC_MUTE_EXCEPTION = "grpcMuteException";
+ /*
+ * The exchange property with an AtomicBoolean, shared by the exchanges of
one call of a consumer with
+ * routeControlledStreamObserver=true, that the
toRouteControlledStreamObserver producer sets once it ended the call
+ * with an error: the producer then ignores the later events of the call,
as nothing more can be sent to the client
+ */
+ String GRPC_RESPONSE_OBSERVER_FAILED = "grpcResponseObserverFailed";
}
diff --git
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java
index efaad79a2ab6..ecde5e83acf2 100644
---
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java
+++
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java
@@ -16,12 +16,22 @@
*/
package org.apache.camel.component.grpc;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
+import org.apache.camel.component.grpc.server.GrpcMethodHandler;
import org.apache.camel.support.DefaultProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class GrpcProducerToRouteControlledStreamObserver extends
DefaultProducer {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(GrpcProducerToRouteControlledStreamObserver.class);
+
public GrpcProducerToRouteControlledStreamObserver(Endpoint endpoint) {
super(endpoint);
}
@@ -30,15 +40,108 @@ public class GrpcProducerToRouteControlledStreamObserver
extends DefaultProducer
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) {
+ // compareAndSet, as a concurrent onNext whose reply does not
convert can end the call at the same time
+ if (failed.compareAndSet(false, true)) {
+ observer.onError(throwable);
+ } else {
+ LOG.debug("Ignoring the {} event of exchange {}: the call
already ended with an error",
+ GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR,
exchange.getExchangeId());
+ }
+ return;
+ }
+ IllegalArgumentException e = new IllegalArgumentException(
+ "The body of the " + GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR
+ + " event
must be a Throwable, but was "
+ + (body ==
null ? "null" : "a " + body.getClass().getName()));
+ if (failed.compareAndSet(false, true)) {
+ String description = exchange.getMessage().getBody(String.class);
+
observer.onError(Status.INTERNAL.withDescription(description).withCause(e).asRuntimeException());
}
+ throw e;
}
}
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 91404959c5f2..c7c1f1b4cb5e 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
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
@@ -75,7 +76,8 @@ public class GrpcMethodHandler {
exchange.getIn().setHeaders(grcpHeaders);
if (endpoint.getConfiguration().isRouteControlledStreamObserver()) {
- exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+ setRouteControlledProperties(endpoint, exchange, responseObserver,
getResponseType(methodName),
+ new AtomicBoolean());
invokeRoute(endpoint, exchange);
return;
}
@@ -90,6 +92,26 @@ public class GrpcMethodHandler {
}
}
+ /**
+ * Sets the exchange properties that the toRouteControlledStreamObserver
producer uses to reply to the client, for a
+ * consumer with routeControlledStreamObserver=true.
+ *
+ * @param endpoint the endpoint of the consumer
+ * @param exchange the exchange of the event of the call
+ * @param responseObserver the response stream observer of the call
+ * @param responseType the response message type of the method, or
null when unknown
+ * @param failed set once the producer ended the call with an
error, the same for all the exchanges of the
+ * call
+ */
+ static void setRouteControlledProperties(
+ GrpcEndpoint endpoint, Exchange exchange, StreamObserver<Object>
responseObserver, Class<?> responseType,
+ AtomicBoolean failed) {
+ exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+ exchange.setProperty(GrpcConstants.GRPC_RESPONSE_TYPE, responseType);
+ exchange.setProperty(GrpcConstants.GRPC_MUTE_EXCEPTION,
endpoint.getConfiguration().isMuteException());
+ exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER_FAILED,
failed);
+ }
+
/**
* Sends the reply of the route to the client: each element of a List
body, or the body itself, converted to the
* response type of the method. When a message is not of the response type
and cannot be converted to it, nothing is
@@ -106,6 +128,30 @@ public class GrpcMethodHandler {
static boolean sendResponses(
GrpcEndpoint endpoint, Exchange exchange, Object body, Class<?>
responseType,
StreamObserver<Object> responseObserver) {
+ List<Object> responses;
+ try {
+ responses = toResponses(exchange, body, responseType);
+ } catch (IllegalArgumentException e) {
+ LOG.warn("Cannot send the reply of exchange {}: {}",
exchange.getExchangeId(), e.getMessage());
+ responseObserver.onError(toStatusException(endpoint, e));
+ return false;
+ }
+ responses.forEach(responseObserver::onNext);
+ return true;
+ }
+
+ /**
+ * The response messages to send to the client: each element of a List
body, or the body itself, converted to the
+ * response type of the method.
+ *
+ * @param exchange the exchange of the reply
+ * @param body the reply of the route
+ * @param responseType the response message type of the
method, or null when unknown to keep the body
+ * as is
+ * @return the response messages
+ * @throws IllegalArgumentException when a message is not of the response
type and cannot be converted to it
+ */
+ public static List<Object> toResponses(Exchange exchange, Object body,
Class<?> responseType) {
List<?> messages = body instanceof List<?> list ? list :
Collections.singletonList(body);
List<Object> responses = new ArrayList<>(messages.size());
for (Object message : messages) {
@@ -117,16 +163,13 @@ public class GrpcMethodHandler {
try {
responses.add(exchange.getContext().getTypeConverter().mandatoryConvertTo(responseType,
exchange, message));
} catch (NoTypeConversionAvailableException |
TypeConversionException e) {
- String reason = "The reply of the gRPC method must be a " +
responseType.getName()
- + " or a List of them, but was a " +
message.getClass().getName();
- IllegalArgumentException cause = new
IllegalArgumentException(reason, e);
- LOG.warn("Cannot send the reply of exchange {}: {}",
exchange.getExchangeId(), cause.getMessage());
- responseObserver.onError(toStatusException(endpoint, cause));
- return false;
+ throw new IllegalArgumentException(
+ "The reply of the gRPC method must be a " +
responseType.getName()
+ + " or a List of them, but
was a " + message.getClass().getName(),
+ e);
}
}
- responses.forEach(responseObserver::onNext);
- return true;
+ return responses;
}
/**
@@ -165,10 +208,20 @@ public class GrpcMethodHandler {
/**
* The error to send to the client when the exchange failed.
*/
- static StatusRuntimeException toStatusException(GrpcEndpoint endpoint,
Exception cause) {
+ public static StatusRuntimeException toStatusException(GrpcEndpoint
endpoint, Exception cause) {
+ return
toStatusException(endpoint.getConfiguration().isMuteException(), cause);
+ }
+
+ /**
+ * The error to send to the client when the exchange failed.
+ *
+ * @param muteException whether the description sent to the client leaves
out the message of the cause
+ * @param cause the cause of the failure
+ */
+ public static StatusRuntimeException toStatusException(boolean
muteException, Exception cause) {
// the description IS transmitted to the client, so it carries the
route's exception message only when
- // the endpoint opted out of muting; the cause below stays local
either way
- String description = endpoint.getConfiguration().isMuteException() ?
MUTED_DESCRIPTION : cause.getMessage();
+ // the consumer opted out of muting; the cause below stays local
either way
+ String description = muteException ? MUTED_DESCRIPTION :
cause.getMessage();
return Status.INTERNAL
.withDescription(description)
// This can be attached to the Status locally, but NOT
transmitted to the client!
diff --git
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestDelegationStreamObserver.java
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestDelegationStreamObserver.java
index bb7d2bd2018b..2459ad4c1115 100644
---
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestDelegationStreamObserver.java
+++
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestDelegationStreamObserver.java
@@ -17,14 +17,18 @@
package org.apache.camel.component.grpc.server;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import io.grpc.stub.StreamObserver;
-import org.apache.camel.component.grpc.GrpcConstants;
+import org.apache.camel.Exchange;
import org.apache.camel.component.grpc.GrpcConsumer;
import org.apache.camel.component.grpc.GrpcEndpoint;
public class GrpcRequestDelegationStreamObserver extends
GrpcRequestAbstractStreamObserver {
+ // set once the route ended the call with an error, shared by the
exchanges of all the events of the call
+ private final AtomicBoolean failed = new AtomicBoolean();
+
public GrpcRequestDelegationStreamObserver(GrpcEndpoint endpoint,
GrpcConsumer consumer,
StreamObserver<Object>
responseObserver, Map<String, Object> headers) {
super(endpoint, consumer, responseObserver, headers);
@@ -39,7 +43,7 @@ public class GrpcRequestDelegationStreamObserver extends
GrpcRequestAbstractStre
var exchange = endpoint.createExchange();
exchange.getIn().setBody(request);
exchange.getIn().setHeaders(headers);
- exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+ setRouteControlledProperties(exchange);
consumer.process(exchange, doneSync -> {
});
}
@@ -48,7 +52,7 @@ public class GrpcRequestDelegationStreamObserver extends
GrpcRequestAbstractStre
public void onError(Throwable throwable) {
var exchange = endpoint.createExchange();
exchange.getIn().setHeaders(headers);
- exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+ setRouteControlledProperties(exchange);
consumer.onError(exchange, throwable);
}
@@ -56,7 +60,11 @@ public class GrpcRequestDelegationStreamObserver extends
GrpcRequestAbstractStre
public void onCompleted() {
var exchange = endpoint.createExchange();
exchange.getIn().setHeaders(headers);
- exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+ setRouteControlledProperties(exchange);
consumer.onCompleted(exchange);
}
+
+ private void setRouteControlledProperties(Exchange exchange) {
+ GrpcMethodHandler.setRouteControlledProperties(endpoint, exchange,
responseObserver, responseType, failed);
+ }
}
diff --git
a/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcRouteControlledReplyTypeTest.java
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcRouteControlledReplyTypeTest.java
new file mode 100644
index 000000000000..d388a40f353d
--- /dev/null
+++
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcRouteControlledReplyTypeTest.java
@@ -0,0 +1,375 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.grpc;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.grpc.GrpcConsumerReplyTypeTest.Pong;
+import
org.apache.camel.component.grpc.GrpcConsumerReplyTypeTest.PongResponseStreamObserver;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeEvent;
+import org.apache.camel.support.EventNotifierSupport;
+import org.apache.camel.support.TypeConverterSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * With routeControlledStreamObserver=true, a reply sent by the
toRouteControlledStreamObserver producer must be of the
+ * response type of the called method: a body of another type is converted to
it, or the call ends with an error. It
+ * must never be serialized as is, as the client would decode the bytes of,
for example, the unchanged request as a
+ * response.
+ */
+public class GrpcRouteControlledReplyTypeTest extends GrpcTestSupport {
+
+ private static final String PING_NAME = "PING";
+ private static final int PING_ID = 7;
+
+ private static final String CONSUMER =
"grpc://localhost:0/org.apache.camel.component.grpc.PingPong?synchronous=true"
+ +
"&routeControlledStreamObserver=true";
+ private static final String REPLY =
"grpc://dummy:0/?toRouteControlledStreamObserver=true";
+
+ private final Map<String, ManagedChannel> channels = new
ConcurrentHashMap<>();
+ // the exception of each exchange done by a consumer route (null when it
completed), by route id
+ private final Map<String, List<Exception>> doneExchanges = new
ConcurrentHashMap<>();
+
+ @AfterEach
+ public void stopGrpcChannels() {
+ channels.values().forEach(channel -> channel.shutdown().shutdownNow());
+ channels.clear();
+ }
+
+ @Test
+ public void unchangedRequestIsNotSentAsResponse() {
+ StatusRuntimeException e = assertThrows(StatusRuntimeException.class,
+ () ->
blockingStub("rc-unchanged").pingSyncSync(pingRequest()));
+
+ // muteException=false of the consumer gives the real description
+ assertReplyTypeError(e.getStatus(), PingRequest.class);
+ assertReplyTypeFailure(awaitDoneExchanges("rc-unchanged", 1).get(0));
+ }
+
+ @Test
+ public void redeliveryFailsWithTheReplyTypeError() {
+ StatusRuntimeException e = assertThrows(StatusRuntimeException.class,
+ () ->
blockingStub("rc-unchanged-redelivery").pingSyncSync(pingRequest()));
+
+ assertReplyTypeError(e.getStatus(), PingRequest.class);
+ // the redelivery does not call the closed call again, so the exchange
fails with the reply type error
+ assertReplyTypeFailure(awaitDoneExchanges("rc-unchanged-redelivery",
1).get(0));
+ }
+
+ @Test
+ public void replyTypeErrorIsMutedByDefault() {
+ StatusRuntimeException e = assertThrows(StatusRuntimeException.class,
+ () ->
blockingStub("rc-unchanged-muted").pingSyncSync(pingRequest()));
+
+ assertEquals(Status.Code.INTERNAL, e.getStatus().getCode());
+ assertEquals("Exchange processing failed",
e.getStatus().getDescription());
+ }
+
+ @Test
+ public void replyIsConvertedToTheResponseType() {
+ PongResponse pongResponse =
blockingStub("rc-convert").pingSyncSync(pingRequest());
+
+ assertEquals(PING_NAME + "PONG", pongResponse.getPongName());
+ assertEquals(PING_ID, pongResponse.getPongId());
+ }
+
+ @Test
+ public void serverStreamingSendsEachMessageOfAList() {
+ Iterator<PongResponse> responses =
blockingStub("rc-streaming-list").pingSyncAsync(pingRequest());
+
+ assertEquals(PING_NAME + "PONG", responses.next().getPongName());
+ assertEquals(PING_NAME + "PONG2", responses.next().getPongName());
+ assertFalse(responses.hasNext());
+ }
+
+ @Test
+ public void delegationUnchangedRequestIsNotSentAsResponse() throws
Exception {
+ PongResponseStreamObserver responseObserver =
callBidiStreaming("rc-delegation-unchanged");
+
+ assertTrue(responseObserver.pongResponses.isEmpty(), "the request must
not be sent as response");
+ StatusRuntimeException e =
assertInstanceOf(StatusRuntimeException.class, responseObserver.error);
+ assertReplyTypeError(e.getStatus(), PingRequest.class);
+ assertEquals(0, responseObserver.completed.get());
+ // the onNext exchange fails; the forwarded onCompleted event is
ignored on the closed call
+ List<Exception> exceptions =
awaitDoneExchanges("rc-delegation-unchanged", 2);
+ assertReplyTypeFailure(exceptions.get(0));
+ assertNull(exceptions.get(1), "the onCompleted exchange must not
fail");
+ }
+
+ @Test
+ public void delegationConvertsTheReply() throws Exception {
+ PongResponseStreamObserver responseObserver =
callBidiStreaming("rc-delegation-convert");
+
+ assertEquals(1, responseObserver.pongResponses.size());
+ assertEquals(PING_NAME + "PONG",
responseObserver.pongResponses.get(0).getPongName());
+ assertEquals(1, responseObserver.completed.get());
+ }
+
+ @Test
+ public void missingEventTypeHeaderFailsTheExchangeAndSendsNothing() {
+ PongResponseStreamObserver responseObserver = new
PongResponseStreamObserver();
+
+ Exchange exchange = template.send(REPLY, e -> {
+ e.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+
e.getMessage().setBody(PongResponse.newBuilder().setPongName(PING_NAME +
"PONG").build());
+ });
+
+ IllegalArgumentException e =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+
assertTrue(e.getMessage().contains(GrpcConstants.GRPC_EVENT_TYPE_HEADER),
e.getMessage());
+
assertTrue(e.getMessage().contains(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT),
e.getMessage());
+
assertTrue(e.getMessage().contains(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED),
e.getMessage());
+
assertTrue(e.getMessage().contains(GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR),
e.getMessage());
+ assertTrue(responseObserver.pongResponses.isEmpty(), "nothing must be
sent to the client");
+ assertNull(responseObserver.error, "nothing must be sent to the
client");
+ assertEquals(0, responseObserver.completed.get());
+ }
+
+ @Test
+ public void missingResponseObserverFailsTheExchange() {
+ Exchange exchange = template.send(REPLY, e -> {
+ e.getMessage().setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT);
+
e.getMessage().setBody(PongResponse.newBuilder().setPongName(PING_NAME +
"PONG").build());
+ });
+
+ IllegalArgumentException e =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+
assertTrue(e.getMessage().contains(GrpcConstants.GRPC_RESPONSE_OBSERVER),
e.getMessage());
+
assertTrue(e.getMessage().contains("routeControlledStreamObserver=true"),
e.getMessage());
+ }
+
+ @Test
+ public void unknownEventTypeFailsTheExchangeAndSendsNothing() {
+ PongResponseStreamObserver responseObserver = new
PongResponseStreamObserver();
+
+ Exchange exchange = template.send(REPLY, e -> {
+ e.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+ e.getMessage().setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
"ON_NEXT");
+
e.getMessage().setBody(PongResponse.newBuilder().setPongName(PING_NAME +
"PONG").build());
+ });
+
+ IllegalArgumentException e =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertTrue(e.getMessage().contains("ON_NEXT"), e.getMessage());
+
assertTrue(e.getMessage().contains(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT + ",
"), e.getMessage());
+ assertTrue(responseObserver.pongResponses.isEmpty(), "nothing must be
sent to the client");
+ assertNull(responseObserver.error, "nothing must be sent to the
client");
+ assertEquals(0, responseObserver.completed.get());
+ }
+
+ @Test
+ public void onErrorWithTextBodyEndsTheCallAndFailsTheExchange() {
+ PongResponseStreamObserver responseObserver = new
PongResponseStreamObserver();
+
+ Exchange exchange = sendOnError(responseObserver, "No pong today");
+
+ IllegalArgumentException e =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertTrue(e.getMessage().contains(String.class.getName()),
e.getMessage());
+ StatusRuntimeException error =
assertInstanceOf(StatusRuntimeException.class, responseObserver.error);
+ assertEquals(Status.Code.INTERNAL, error.getStatus().getCode());
+ assertEquals("No pong today", error.getStatus().getDescription());
+ }
+
+ @Test
+ public void onErrorWithoutBodyEndsTheCallAndFailsTheExchange() {
+ PongResponseStreamObserver responseObserver = new
PongResponseStreamObserver();
+
+ Exchange exchange = sendOnError(responseObserver, null);
+
+ IllegalArgumentException e =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertTrue(e.getMessage().contains("null"), e.getMessage());
+ StatusRuntimeException error =
assertInstanceOf(StatusRuntimeException.class, responseObserver.error);
+ assertEquals(Status.Code.INTERNAL, error.getStatus().getCode());
+ assertNull(error.getStatus().getDescription());
+ }
+
+ private Exchange sendOnError(PongResponseStreamObserver responseObserver,
Object body) {
+ return template.send(REPLY, e -> {
+ e.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+ e.getMessage().setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR);
+ e.getMessage().setBody(body);
+ });
+ }
+
+ private List<Exception> awaitDoneExchanges(String routeId, int count) {
+ List<Exception> exceptions
+ = doneExchanges.computeIfAbsent(routeId, id ->
Collections.synchronizedList(new ArrayList<>()));
+ await().atMost(5, TimeUnit.SECONDS).until(() -> exceptions.size() >=
count);
+ synchronized (exceptions) {
+ assertEquals(count, exceptions.size(), "exchanges done by route "
+ routeId + ": " + exceptions);
+ return new ArrayList<>(exceptions);
+ }
+ }
+
+ private static void assertReplyTypeFailure(Exception exception) {
+ IllegalArgumentException e =
assertInstanceOf(IllegalArgumentException.class, exception);
+ assertTrue(e.getMessage().contains(PongResponse.class.getName()),
e.getMessage());
+ assertTrue(e.getMessage().contains(PingRequest.class.getName()),
e.getMessage());
+ }
+
+ private static void assertReplyTypeError(Status status, Class<?>
replyType) {
+ assertEquals(Status.Code.INTERNAL, status.getCode());
+
assertTrue(status.getDescription().contains(PongResponse.class.getName()),
+ "the description must name the response type: " + status);
+ assertTrue(status.getDescription().contains(replyType.getName()),
+ "the description must name the type of the reply: " + status);
+ }
+
+ private static PingRequest pingRequest() {
+ return
PingRequest.newBuilder().setPingName(PING_NAME).setPingId(PING_ID).build();
+ }
+
+ private ManagedChannel channel(String routeId) {
+ return channels.computeIfAbsent(routeId,
+ id -> ManagedChannelBuilder.forAddress("localhost",
getRoutePort(id)).usePlaintext().build());
+ }
+
+ private PingPongGrpc.PingPongBlockingStub blockingStub(String routeId) {
+ return
PingPongGrpc.newBlockingStub(channel(routeId)).withDeadlineAfter(5,
TimeUnit.SECONDS);
+ }
+
+ private PongResponseStreamObserver callBidiStreaming(String routeId)
throws InterruptedException {
+ PongResponseStreamObserver responseObserver = new
PongResponseStreamObserver();
+ StreamObserver<PingRequest> requestObserver =
PingPongGrpc.newStub(channel(routeId)).pingAsyncAsync(responseObserver);
+ requestObserver.onNext(pingRequest());
+ requestObserver.onCompleted();
+ assertTrue(responseObserver.latch.await(5, TimeUnit.SECONDS));
+ return responseObserver;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ getContext().getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ Exchange exchange = ((ExchangeEvent)
event).getExchange();
+ if (exchange.getFromRouteId() != null) {
+
doneExchanges.computeIfAbsent(exchange.getFromRouteId(),
+ id -> Collections.synchronizedList(new
ArrayList<>()))
+ .add(exchange.getException());
+ }
+ }
+
+ @Override
+ public boolean isEnabled(CamelEvent event) {
+ return event instanceof
CamelEvent.ExchangeCompletedEvent
+ || event instanceof
CamelEvent.ExchangeFailedEvent;
+ }
+ });
+
+
getContext().getTypeConverterRegistry().addTypeConverter(PongResponse.class,
Pong.class,
+ new TypeConverterSupport() {
+ @Override
+ public <T> T convertTo(Class<T> type, Exchange
exchange, Object value) {
+ Pong pong = (Pong) value;
+ return
type.cast(PongResponse.newBuilder().setPongName(pong.name()).setPongId(pong.id())
+ .build());
+ }
+ });
+
+ // unary and server streaming calls: the route sends the
reply, then completes the call
+ from(CONSUMER + "&hash=1&muteException=false")
+ .routeId("rc-unchanged")
+ .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
constant(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT))
+ .to(REPLY)
+ .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
+
constant(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED))
+ .to(REPLY);
+
+ from(CONSUMER + "&hash=2")
+ .routeId("rc-unchanged-muted")
+ .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
constant(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT))
+ .to(REPLY)
+ .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
+
constant(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED))
+ .to(REPLY);
+
+ from(CONSUMER + "&hash=3")
+ .routeId("rc-convert")
+ .process(e ->
e.getMessage().setBody(toPong(e.getMessage().getBody(PingRequest.class),
"PONG")))
+ .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
constant(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT))
+ .to(REPLY)
+ .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
+
constant(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED))
+ .to(REPLY);
+
+ from(CONSUMER + "&hash=4")
+ .routeId("rc-streaming-list")
+ .process(e -> {
+ PingRequest request =
e.getMessage().getBody(PingRequest.class);
+ e.getMessage().setBody(List.of(toPong(request,
"PONG"), toPong(request, "PONG2")));
+ })
+ .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
constant(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT))
+ .to(REPLY)
+ .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
+
constant(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED))
+ .to(REPLY);
+
+ from(CONSUMER + "&hash=7&muteException=false")
+ .routeId("rc-unchanged-redelivery")
+
.errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(0))
+ .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
constant(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT))
+ .to(REPLY)
+ .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER,
+
constant(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED))
+ .to(REPLY);
+
+ // client and bidirectional streaming calls: the consumer
forwards each event with its type header
+ from(CONSUMER +
"&hash=5&consumerStrategy=DELEGATION&forwardOnCompleted=true&muteException=false")
+ .routeId("rc-delegation-unchanged")
+ .to(REPLY);
+
+ from(CONSUMER +
"&hash=6&consumerStrategy=DELEGATION&forwardOnCompleted=true")
+ .routeId("rc-delegation-convert")
+ .process(e -> {
+ PingRequest request =
e.getMessage().getBody(PingRequest.class);
+ if (request != null) {
+ e.getMessage().setBody(toPong(request,
"PONG"));
+ }
+ })
+ .to(REPLY);
+ }
+ };
+ }
+
+ private static Pong toPong(PingRequest request, String suffix) {
+ return new Pong(request.getPingName() + suffix, request.getPingId());
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index ba7d3a6fc0f5..6853e0c9fed9 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1043,6 +1043,14 @@ of the called method, and when that is not possible it
sends nothing and ends th
A route must set the response message of the method as the body. A reply
without a body still sends an empty response
message.
+The same applies to the replies that a producer with
`toRouteControlledStreamObserver=true` sends for a consumer with
+`routeControlledStreamObserver=true`: a reply that does not convert ends the
call with an `INTERNAL` status error and
+fails the exchange. Such a producer also sends each element of a `List` body
as a message, where a `List` failed
+before. A `CamelGrpcEventType` header with a value other than `onNext`,
`onCompleted` or `onError` now fails the
+exchange, where it sent nothing before. An `onError` event with a body that is
not a `Throwable` now ends the call with
+an `INTERNAL` status error and fails the exchange with an
`IllegalArgumentException`, where it failed with a
+`ClassCastException` or a `NullPointerException` and the client waited until
its deadline.
+
=== camel-hashicorp-vault - cloud without a namespace, the createSecret body,
and the secret version
With `cloud=true` and no `namespace`, the `hashicorp-vault` producer now works
on the secrets engine itself, as the
