This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26187-grpc-reply-type in repository https://gitbox.apache.org/repos/asf/camel.git
commit e025ec9c3832850dba89975ae9cd9e5700bc98be Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 23:53:59 2026 +0100 CAMEL-26187: camel-grpc - the consumer converts the reply to the response 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 10f5a47c1b89..1a76d1f758bb 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 @@ -788,6 +788,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
