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 6a7e17a1af69 CAMEL-26187: camel-grpc - the consumer converts the reply
to the response type of the method, or fails the call (#29012)
6a7e17a1af69 is described below
commit 6a7e17a1af69408bc004b8a0653ff7623c41639f
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 11:42:18 2026 +0100
CAMEL-26187: camel-grpc - the consumer converts the reply to the response
type of the method, or fails the call (#29012)
…onse type of the method, or fails the call
The consumer sent the body at the end of the route to the client as is.
When the
route left the request unchanged, the PingRequest was serialized and the
client
decoded the bytes as a PongResponse: no error, wrong field values.
The consumer now looks up the response message type of the called method
(the
generic type of its response StreamObserver) and converts each reply
message to
it with a type converter. When that is not possible it sends nothing and
ends the
call with an INTERNAL status error (muted by muteException). This applies
to unary
and server streaming calls and to the AGGREGATION and PROPAGATION
strategies.
No body still sends an empty message.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../org/apache/camel/catalog/components/grpc.json | 2 +-
.../apache/camel/catalog/docs/grpc-component.adoc | 10 +
.../org/apache/camel/component/grpc/grpc.json | 2 +-
.../camel-grpc/src/main/docs/grpc-component.adoc | 10 +
.../apache/camel/component/grpc/GrpcEndpoint.java | 6 +-
.../component/grpc/server/GrpcMethodHandler.java | 101 +++++++-
.../server/GrpcRequestAbstractStreamObserver.java | 23 ++
.../GrpcRequestAggregationStreamObserver.java | 9 +-
.../GrpcRequestPropagationStreamObserver.java | 9 +-
.../component/grpc/GrpcConsumerReplyTypeTest.java | 283 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 +
11 files changed, 438 insertions(+), 26 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/grpc.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/grpc.json
index 4d235ea68662..b474fa62a1a1 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/grpc.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/grpc.json
@@ -36,7 +36,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "pojo" ], "byOption": [ { "option":
"consumerStrategy", "value": "AGGREGATION", "carriers": [ "pojo", "list" ] }, {
"option": "forwardOnCompleted", "value": "true", "carriers": [ "pojo", "null" ]
} ], "description": "The protobuf request message of the called method, with
the method name in the CamelGrpcMethodName header. A streaming request gives
one exchange per message, or with the AGGREGATION strategy a List of all its
messages. forwardOnCompleted s [...]
- "consumerReply": { "carriers": [ "pojo", "list", "null" ], "dependsOn": [
"routeControlledStreamObserver" ], "description": "The protobuf response
message, or a List of them for a streaming response; no body sends an empty
message. Text, bytes or a Map fail: build the message of the method. With
routeControlledStreamObserver the route replies through a producer with
toRouteControlledStreamObserver" },
+ "consumerReply": { "carriers": [ "pojo", "list", "null" ], "dependsOn": [
"routeControlledStreamObserver" ], "description": "The protobuf response
message of the method, or a List of them for a streaming response; no body
sends an empty message. Another type is converted to the response message, or
the call ends with an INTERNAL error: text, bytes, a Map or the unchanged
request fail, build the message of the method. With
routeControlledStreamObserver the route replies through a prod [...]
"producerIn": { "carriers": [ "pojo", "list" ], "description": "The
protobuf request message of the method, or a List of them for a streaming
request. Text, bytes or a Map fail with an argument type mismatch: build the
message first" },
"producerOut": { "carriers": [ "list" ], "byOption": [ { "option":
"synchronous", "value": "true", "carriers": [ "pojo", "list" ] }, { "option":
"producerStrategy", "value": "STREAMING", "carriers": [ "unchanged" ] }, {
"option": "toRouteControlledStreamObserver", "value": "true", "carriers": [
"unchanged" ] } ], "description": "A List of the protobuf response messages.
With synchronous=true the response message itself, or a List of them for a
streaming response. With the STREAMING s [...]
},
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 c092f6a8e8ed..9cc188ba41a3 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
@@ -73,6 +73,16 @@ To enable these features, the following component properties
combinations must b
|||jwtSubject||Optional
|=======================================================================
+=== gRPC consumer reply
+
+The message body at the end of the consumer route is the reply to the client:
the response message of the called
+method, or a `List` of them for a streaming response. A message without a body
sends an empty response message.
+
+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.
+
=== gRPC producer resource type mapping
The table below shows the types of objects in the message body, depending on
the types (simple or stream) of incoming and outgoing parameters, as well as
the invocation style (synchronous or asynchronous). Please note that invocation
of the procedures with incoming stream parameter in asynchronous style is not
allowed.
diff --git
a/components/camel-grpc/src/generated/resources/META-INF/org/apache/camel/component/grpc/grpc.json
b/components/camel-grpc/src/generated/resources/META-INF/org/apache/camel/component/grpc/grpc.json
index 4d235ea68662..b474fa62a1a1 100644
---
a/components/camel-grpc/src/generated/resources/META-INF/org/apache/camel/component/grpc/grpc.json
+++
b/components/camel-grpc/src/generated/resources/META-INF/org/apache/camel/component/grpc/grpc.json
@@ -36,7 +36,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "pojo" ], "byOption": [ { "option":
"consumerStrategy", "value": "AGGREGATION", "carriers": [ "pojo", "list" ] }, {
"option": "forwardOnCompleted", "value": "true", "carriers": [ "pojo", "null" ]
} ], "description": "The protobuf request message of the called method, with
the method name in the CamelGrpcMethodName header. A streaming request gives
one exchange per message, or with the AGGREGATION strategy a List of all its
messages. forwardOnCompleted s [...]
- "consumerReply": { "carriers": [ "pojo", "list", "null" ], "dependsOn": [
"routeControlledStreamObserver" ], "description": "The protobuf response
message, or a List of them for a streaming response; no body sends an empty
message. Text, bytes or a Map fail: build the message of the method. With
routeControlledStreamObserver the route replies through a producer with
toRouteControlledStreamObserver" },
+ "consumerReply": { "carriers": [ "pojo", "list", "null" ], "dependsOn": [
"routeControlledStreamObserver" ], "description": "The protobuf response
message of the method, or a List of them for a streaming response; no body
sends an empty message. Another type is converted to the response message, or
the call ends with an INTERNAL error: text, bytes, a Map or the unchanged
request fail, build the message of the method. With
routeControlledStreamObserver the route replies through a prod [...]
"producerIn": { "carriers": [ "pojo", "list" ], "description": "The
protobuf request message of the method, or a List of them for a streaming
request. Text, bytes or a Map fail with an argument type mismatch: build the
message first" },
"producerOut": { "carriers": [ "list" ], "byOption": [ { "option":
"synchronous", "value": "true", "carriers": [ "pojo", "list" ] }, { "option":
"producerStrategy", "value": "STREAMING", "carriers": [ "unchanged" ] }, {
"option": "toRouteControlledStreamObserver", "value": "true", "carriers": [
"unchanged" ] } ], "description": "A List of the protobuf response messages.
With synchronous=true the response message itself, or a List of them for a
streaming response. With the STREAMING s [...]
},
diff --git a/components/camel-grpc/src/main/docs/grpc-component.adoc
b/components/camel-grpc/src/main/docs/grpc-component.adoc
index c092f6a8e8ed..9cc188ba41a3 100644
--- a/components/camel-grpc/src/main/docs/grpc-component.adoc
+++ b/components/camel-grpc/src/main/docs/grpc-component.adoc
@@ -73,6 +73,16 @@ To enable these features, the following component properties
combinations must b
|||jwtSubject||Optional
|=======================================================================
+=== gRPC consumer reply
+
+The message body at the end of the consumer route is the reply to the client:
the response message of the called
+method, or a `List` of them for a streaming response. A message without a body
sends an empty response message.
+
+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.
+
=== gRPC producer resource type mapping
The table below shows the types of objects in the message body, depending on
the types (simple or stream) of incoming and outgoing parameters, as well as
the invocation style (synchronous or asynchronous). Please note that invocation
of the procedures with incoming stream parameter in asynchronous style is not
allowed.
diff --git
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcEndpoint.java
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcEndpoint.java
index ec7ace7f6be5..2a7a389e4d7b 100644
---
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcEndpoint.java
+++
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcEndpoint.java
@@ -51,8 +51,10 @@ import org.apache.camel.util.ObjectHelper;
+ " the Throwable"),
consumerReply = @BodyType(value = { Carrier.POJO, Carrier.LIST,
Carrier.NULL },
dependsOn =
"routeControlledStreamObserver",
- description = "The protobuf response
message, or a List of them for a streaming response;"
- + " no body sends an empty
message. Text, bytes or a Map fail: build the"
+ description = "The protobuf response
message of the method, or a List of them for a"
+ + " streaming response; no
body sends an empty message. Another type is"
+ + " converted to the
response message, or the call ends with an INTERNAL"
+ + " error: text, bytes, a
Map or the unchanged request fail, build the"
+ " message of the method.
With routeControlledStreamObserver the route"
+ " replies through a
producer with toRouteControlledStreamObserver"),
producerIn = @BodyType(value = { Carrier.POJO, Carrier.LIST },
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 da5b410d8616..91404959c5f2 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
@@ -16,27 +16,43 @@
*/
package org.apache.camel.component.grpc.server;
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import org.apache.camel.Exchange;
+import org.apache.camel.NoTypeConversionAvailableException;
+import org.apache.camel.TypeConversionException;
import org.apache.camel.component.grpc.GrpcConstants;
import org.apache.camel.component.grpc.GrpcConsumer;
import org.apache.camel.component.grpc.GrpcConsumerStrategy;
import org.apache.camel.component.grpc.GrpcEndpoint;
+import org.apache.camel.component.grpc.GrpcUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Handles gRPC service method invocations
*/
public class GrpcMethodHandler {
+ private static final Logger LOG =
LoggerFactory.getLogger(GrpcMethodHandler.class);
+
private static final String MUTED_DESCRIPTION = "Exchange processing
failed";
protected final GrpcConsumer consumer;
+ // the response message type of each method of the service, by method name
(empty when unknown)
+ private final Map<String, Optional<Class<?>>> responseTypes = new
ConcurrentHashMap<>();
public GrpcMethodHandler(GrpcConsumer consumer) {
this.consumer = consumer;
@@ -68,18 +84,84 @@ public class GrpcMethodHandler {
if (exchange.isFailed()) {
responseObserver.onError(toStatusException(endpoint,
exchange.getException()));
- } else {
- Object responseBody = exchange.getIn().getBody();
- if (responseBody instanceof List) {
- List<Object> responseList = (List<Object>) responseBody;
- responseList.forEach(responseObserver::onNext);
- } else {
- responseObserver.onNext(responseBody);
- }
+ } else if (sendResponses(endpoint, exchange,
exchange.getIn().getBody(), getResponseType(methodName),
+ responseObserver)) {
responseObserver.onCompleted();
}
}
+ /**
+ * 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
+ * sent and the call ends with an error, as the client would otherwise
decode the bytes of another message type (for
+ * example the unchanged request) as the response.
+ *
+ * @param endpoint the endpoint
+ * @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 send the body as is
+ * @param responseObserver the response stream observer
+ * @return true if the reply was sent, false if the call
ended with an error
+ */
+ static boolean sendResponses(
+ GrpcEndpoint endpoint, Exchange exchange, Object body, Class<?>
responseType,
+ StreamObserver<Object> responseObserver) {
+ List<?> messages = body instanceof List<?> list ? list :
Collections.singletonList(body);
+ List<Object> responses = new ArrayList<>(messages.size());
+ for (Object message : messages) {
+ // no body sends an empty message
+ if (message == null || responseType == null ||
responseType.isInstance(message)) {
+ responses.add(message);
+ continue;
+ }
+ 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;
+ }
+ }
+ responses.forEach(responseObserver::onNext);
+ return true;
+ }
+
+ /**
+ * The response message type of the method of the service, from the
generic type of its response StreamObserver
+ * parameter.
+ *
+ * @param methodName the name of the method
+ * @return the response type, or null when it cannot be
determined
+ */
+ Class<?> getResponseType(String methodName) {
+ return responseTypes.computeIfAbsent(methodName,
this::findResponseType).orElse(null);
+ }
+
+ private Optional<Class<?>> findResponseType(String methodName) {
+ GrpcEndpoint endpoint = (GrpcEndpoint) consumer.getEndpoint();
+ Class<?> implBase;
+ try {
+ implBase =
GrpcUtils.constructGrpcImplBaseClass(endpoint.getServicePackage(),
endpoint.getServiceName(),
+ endpoint.getCamelContext());
+ } catch (IllegalArgumentException e) {
+ LOG.debug("Cannot determine the response type of the gRPC method
{}: {}", methodName, e.getMessage());
+ return Optional.empty();
+ }
+ for (Method method : implBase.getMethods()) {
+ Type[] parameterTypes = method.getGenericParameterTypes();
+ if (method.getName().equals(methodName) && parameterTypes.length > 0
+ && parameterTypes[parameterTypes.length - 1] instanceof
ParameterizedType observerType
+ && observerType.getRawType() == StreamObserver.class
+ && observerType.getActualTypeArguments()[0] instanceof
Class<?> responseType) {
+ return Optional.of(responseType);
+ }
+ }
+ return Optional.empty();
+ }
+
/**
* The error to send to the client when the exchange failed.
*/
@@ -124,6 +206,9 @@ public class GrpcMethodHandler {
throw new IllegalArgumentException(
"gRPC processing strategy not implemented " +
consumer.getConfiguration().getConsumerStrategy());
}
+ if (requestObserver instanceof GrpcRequestAbstractStreamObserver
observer) {
+ observer.setResponseType(getResponseType(methodName));
+ }
return requestObserver;
}
diff --git
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestAbstractStreamObserver.java
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestAbstractStreamObserver.java
index adb2c6f8827c..0e3de80137db 100644
---
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestAbstractStreamObserver.java
+++
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestAbstractStreamObserver.java
@@ -32,6 +32,7 @@ public abstract class GrpcRequestAbstractStreamObserver
implements StreamObserve
protected Exchange exchange;
protected StreamObserver<Object> responseObserver;
protected Map<String, Object> headers;
+ protected Class<?> responseType;
protected GrpcRequestAbstractStreamObserver(GrpcEndpoint endpoint,
GrpcConsumer consumer,
StreamObserver<Object>
responseObserver, Map<String, Object> headers) {
@@ -41,6 +42,28 @@ public abstract class GrpcRequestAbstractStreamObserver
implements StreamObserve
this.headers = headers;
}
+ /**
+ * The response message type of the method: a reply of another type is
converted to it, or ends the call with an
+ * error.
+ */
+ public Class<?> getResponseType() {
+ return responseType;
+ }
+
+ public void setResponseType(Class<?> responseType) {
+ this.responseType = responseType;
+ }
+
+ /**
+ * Sends the reply of the exchange to the client, converted to the
response type of the method.
+ *
+ * @return true if the reply was sent, false if it is not of the response
type and the call ended with an error
+ */
+ protected boolean sendResponses(Exchange exchange) {
+ return GrpcMethodHandler.sendResponses(endpoint, exchange,
exchange.getMessage().getBody(), responseType,
+ responseObserver);
+ }
+
/**
* Sends the failure of the exchange to the client as an error, like for
unary calls.
*
diff --git
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestAggregationStreamObserver.java
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestAggregationStreamObserver.java
index a56c4b5deaad..0e41a433d75d 100644
---
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestAggregationStreamObserver.java
+++
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestAggregationStreamObserver.java
@@ -66,14 +66,9 @@ public class GrpcRequestAggregationStreamObserver extends
GrpcRequestAbstractStr
return;
}
- Object responseBody = exchange.getMessage().getBody();
- if (responseBody instanceof List) {
- List<?> responseList = (List<?>) responseBody;
- responseList.forEach(responseObserver::onNext);
- } else {
- responseObserver.onNext(responseBody);
+ if (sendResponses(exchange)) {
+ responseObserver.onCompleted();
}
- responseObserver.onCompleted();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
responseObserver.onError(e);
diff --git
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestPropagationStreamObserver.java
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestPropagationStreamObserver.java
index a9e29cc114d9..eaa49b450fa2 100644
---
a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestPropagationStreamObserver.java
+++
b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestPropagationStreamObserver.java
@@ -16,7 +16,6 @@
*/
package org.apache.camel.component.grpc.server;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
@@ -59,12 +58,8 @@ public class GrpcRequestPropagationStreamObserver extends
GrpcRequestAbstractStr
return;
}
- Object responseBody = exchange.getMessage().getBody();
- if (responseBody instanceof List) {
- List<?> responseList = (List<?>) responseBody;
- responseList.forEach(responseObserver::onNext);
- } else {
- responseObserver.onNext(responseBody);
+ if (!sendResponses(exchange)) {
+ failed = true;
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
diff --git
a/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcConsumerReplyTypeTest.java
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcConsumerReplyTypeTest.java
new file mode 100644
index 000000000000..2c2aadfbbfb1
--- /dev/null
+++
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcConsumerReplyTypeTest.java
@@ -0,0 +1,283 @@
+/*
+ * 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.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+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.support.TypeConverterSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+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.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The reply of the consumer 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 GrpcConsumerReplyTypeTest extends GrpcTestSupport {
+
+ private static final String PING_NAME = "PING";
+ private static final int PING_ID = 7;
+
+ private final Map<String, ManagedChannel> channels = 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("grpc-unchanged").pingSyncSync(pingRequest(PING_ID)));
+
+ assertReplyTypeError(e.getStatus(), PingRequest.class);
+ }
+
+ @Test
+ public void textReplyEndsTheCallWithAnError() {
+ StatusRuntimeException e = assertThrows(StatusRuntimeException.class,
+ () ->
blockingStub("grpc-text").pingSyncSync(pingRequest(PING_ID)));
+
+ assertReplyTypeError(e.getStatus(), String.class);
+ }
+
+ @Test
+ public void replyTypeErrorIsMutedByDefault() {
+ StatusRuntimeException e = assertThrows(StatusRuntimeException.class,
+ () ->
blockingStub("grpc-unchanged-muted").pingSyncSync(pingRequest(PING_ID)));
+
+ assertEquals(Status.Code.INTERNAL, e.getStatus().getCode());
+ assertEquals("Exchange processing failed",
e.getStatus().getDescription());
+ }
+
+ @Test
+ public void replyIsConvertedToTheResponseType() {
+ PongResponse pongResponse =
blockingStub("grpc-convert").pingSyncSync(pingRequest(PING_ID));
+
+ assertEquals(PING_NAME + "PONG", pongResponse.getPongName());
+ assertEquals(PING_ID, pongResponse.getPongId());
+ }
+
+ @Test
+ public void serverStreamingSendsNothingWhenOneMessageHasTheWrongType() {
+ Iterator<PongResponse> responses =
blockingStub("grpc-streaming-mixed").pingSyncAsync(pingRequest(PING_ID));
+
+ // the first message is a valid PongResponse, but the whole reply is
checked before any message is sent
+ StatusRuntimeException e = assertThrows(StatusRuntimeException.class,
responses::hasNext);
+ assertReplyTypeError(e.getStatus(), PingRequest.class);
+ }
+
+ @Test
+ public void serverStreamingConvertsEachMessage() {
+ Iterator<PongResponse> responses =
blockingStub("grpc-streaming-convert").pingSyncAsync(pingRequest(PING_ID));
+
+ assertEquals(PING_NAME + "PONG", responses.next().getPongName());
+ assertEquals(PING_NAME + "PONG2", responses.next().getPongName());
+ assertFalse(responses.hasNext());
+ }
+
+ @Test
+ public void aggregationUnchangedRequestsAreNotSentAsResponse() throws
Exception {
+ PongResponseStreamObserver responseObserver =
callStreaming("grpc-aggregation-unchanged", false);
+
+ assertTrue(responseObserver.pongResponses.isEmpty(), "the requests
must not be sent as responses");
+ StatusRuntimeException e =
assertInstanceOf(StatusRuntimeException.class, responseObserver.error);
+ assertReplyTypeError(e.getStatus(), PingRequest.class);
+ assertEquals(0, responseObserver.completed.get());
+ }
+
+ @Test
+ public void propagationUnchangedRequestIsNotSentAsResponse() throws
Exception {
+ PongResponseStreamObserver responseObserver =
callStreaming("grpc-propagation-unchanged", true);
+
+ 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());
+ }
+
+ @Test
+ public void propagationConvertsTheReply() throws Exception {
+ PongResponseStreamObserver responseObserver =
callStreaming("grpc-propagation-convert", true);
+
+ assertEquals(1, responseObserver.pongResponses.size());
+ assertEquals(PING_NAME + "PONG",
responseObserver.pongResponses.get(0).getPongName());
+ assertEquals(1, responseObserver.completed.get());
+ }
+
+ @Test
+ public void noBodySendsAnEmptyMessageAsBefore() {
+ PongResponse pongResponse =
blockingStub("grpc-null").pingSyncSync(pingRequest(PING_ID));
+
+ assertEquals(PongResponse.getDefaultInstance(), pongResponse);
+ }
+
+ 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(int id) {
+ return
PingRequest.newBuilder().setPingName(PING_NAME).setPingId(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 callStreaming(String routeId, boolean
bidi) throws InterruptedException {
+ PingPongGrpc.PingPongStub stub =
PingPongGrpc.newStub(channel(routeId));
+ PongResponseStreamObserver responseObserver = new
PongResponseStreamObserver();
+ StreamObserver<PingRequest> requestObserver
+ = bidi ? stub.pingAsyncAsync(responseObserver) :
stub.pingAsyncSync(responseObserver);
+ requestObserver.onNext(pingRequest(PING_ID));
+ requestObserver.onCompleted();
+ assertTrue(responseObserver.latch.await(5, TimeUnit.SECONDS));
+ return responseObserver;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
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());
+ }
+ });
+
+
from("grpc://localhost:0/org.apache.camel.component.grpc.PingPong?hash=1&synchronous=true&muteException=false")
+ .routeId("grpc-unchanged")
+ .log("${body}");
+
+
from("grpc://localhost:0/org.apache.camel.component.grpc.PingPong?hash=2&synchronous=true")
+ .routeId("grpc-unchanged-muted")
+ .log("${body}");
+
+
from("grpc://localhost:0/org.apache.camel.component.grpc.PingPong?hash=3&synchronous=true&muteException=false")
+ .routeId("grpc-text")
+ .setBody(constant("PONG"));
+
+
from("grpc://localhost:0/org.apache.camel.component.grpc.PingPong?hash=4&synchronous=true&muteException=false")
+ .routeId("grpc-null")
+ .process(e -> e.getMessage().setBody(null));
+
+
from("grpc://localhost:0/org.apache.camel.component.grpc.PingPong?hash=5&synchronous=true")
+ .routeId("grpc-convert")
+ .process(e ->
e.getMessage().setBody(toPong(e.getMessage().getBody(PingRequest.class),
"PONG")));
+
+
from("grpc://localhost:0/org.apache.camel.component.grpc.PingPong?hash=6&synchronous=true&muteException=false")
+ .routeId("grpc-streaming-mixed")
+ .process(e -> {
+ PingRequest request =
e.getMessage().getBody(PingRequest.class);
+ PongResponse pong =
PongResponse.newBuilder().setPongName("PONG").build();
+ e.getMessage().setBody(List.of(pong, request));
+ });
+
+
from("grpc://localhost:0/org.apache.camel.component.grpc.PingPong?hash=7&synchronous=true")
+ .routeId("grpc-streaming-convert")
+ .process(e -> {
+ PingRequest request =
e.getMessage().getBody(PingRequest.class);
+ e.getMessage().setBody(List.of(toPong(request,
"PONG"), toPong(request, "PONG2")));
+ });
+
+
from("grpc://localhost:0/org.apache.camel.component.grpc.PingPong?hash=8&synchronous=true"
+ + "&consumerStrategy=AGGREGATION&muteException=false")
+ .routeId("grpc-aggregation-unchanged")
+ .log("${body}");
+
+
from("grpc://localhost:0/org.apache.camel.component.grpc.PingPong?hash=9&synchronous=true"
+ + "&consumerStrategy=PROPAGATION&muteException=false")
+ .routeId("grpc-propagation-unchanged")
+ .log("${body}");
+
+
from("grpc://localhost:0/org.apache.camel.component.grpc.PingPong?hash=10&synchronous=true"
+ + "&consumerStrategy=PROPAGATION")
+ .routeId("grpc-propagation-convert")
+ .process(e ->
e.getMessage().setBody(toPong(e.getMessage().getBody(PingRequest.class),
"PONG")));
+ }
+ };
+ }
+
+ private static Pong toPong(PingRequest request, String suffix) {
+ return new Pong(request.getPingName() + suffix, request.getPingId());
+ }
+
+ /**
+ * A reply of the route that is not a protobuf message, with a type
converter to PongResponse.
+ */
+ public record Pong(String name, int id) {
+ }
+
+ static class PongResponseStreamObserver implements
StreamObserver<PongResponse> {
+ final List<PongResponse> pongResponses = new CopyOnWriteArrayList<>();
+ final AtomicInteger completed = new AtomicInteger();
+ final CountDownLatch latch = new CountDownLatch(1);
+ volatile Throwable error;
+
+ @Override
+ public void onNext(PongResponse value) {
+ pongResponses.add(value);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ error = t;
+ latch.countDown();
+ }
+
+ @Override
+ public void onCompleted() {
+ completed.incrementAndGet();
+ latch.countDown();
+ }
+ }
+}
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 41f803bb6a4a..b6c15cef76e3 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
@@ -884,6 +884,15 @@ configuration (the `camel.component.google-mail.*`
options), kept until the comp
client the component had already created, for example the one of a
`google-mail:messages/modify` endpoint, so
credentials set only on an endpoint URI no longer reach the transformer: set
them on the component as well.
+=== camel-grpc - the consumer reply must be the response message of the method
+
+The `grpc` consumer sent the message body at the end of the route to the
client as is. A body of another protobuf
+message type, such as the unchanged request, was serialized and decoded by the
client as the response, with wrong field
+values and no error. The consumer now converts the reply (each element of a
`List` reply) to the response message type
+of the called method, and when that is not possible it sends nothing and ends
the call with an `INTERNAL` status error.
+A route must set the response message of the method as the body. A reply
without a body still sends an empty response
+message.
+
=== camel-hivemq - a body that does not convert to a byte array fails the
exchange
The `hivemq` producer published a message body that has no conversion to
`byte[]`, such as a `Map`, a POJO or a
