gnodet-bot commented on code in PR #29300: URL: https://github.com/apache/camel/pull/29300#discussion_r4510837766
########## components/camel-grpc/src/main/docs/grpc-component.adoc: ########## @@ -73,6 +73,37 @@ 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. + +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 Review Comment: 🟡 **Doc/code mismatch: type-conversion behavior not implemented** This sentence claims the `toRouteControlledStreamObserver` producer applies type-conversion to each `onNext` body. However, `GrpcProducerToRouteControlledStreamObserver.onNext()` calls `responseObserver.onNext(body)` directly — no type converter is invoked. Either the conversion needs to be implemented, or this paragraph needs to be corrected. ########## components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcPropagationRouteControlledTest.java: ########## @@ -0,0 +1,178 @@ +/* + * 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.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.stub.StreamObserver; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.grpc.GrpcConsumerReplyTypeTest.PongResponseStreamObserver; Review Comment: 🔴 **Compile error — missing class `GrpcConsumerReplyTypeTest`** This import references `GrpcConsumerReplyTypeTest.PongResponseStreamObserver`, but `GrpcConsumerReplyTypeTest` does not exist in this branch. The PR will not compile. ########## components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcMethodHandler.java: ########## @@ -95,6 +97,31 @@ private void invokeRoute(GrpcEndpoint endpoint, Exchange exchange) throws Except } } + /** + * Sets the exchange properties that the toRouteControlledStreamObserver producer uses to reply to the client, for a + * consumer with routeControlledStreamObserver=true. + */ + 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); + } + + /** + * The error to send to the client when the exchange failed. + */ + public static StatusRuntimeException toStatusException(GrpcEndpoint endpoint, Exception cause) { + String description = endpoint.getConfiguration().isMuteException() + ? MUTED_DESCRIPTION : cause.getMessage(); Review Comment: 🟠**`cause.getMessage()` called without null guard** If `cause` is `null` (passed from `sendFailure()` when `exchange.getException()` returns `null` for a fault message), this line throws a `NullPointerException`. Add a null check: `cause != null ? cause.getMessage() : "Exchange failed"`. ########## components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestAbstractStreamObserver.java: ########## @@ -40,4 +42,45 @@ protected GrpcRequestAbstractStreamObserver(GrpcEndpoint endpoint, GrpcConsumer this.responseObserver = responseObserver; 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) { + Object responseBody = exchange.getMessage().getBody(); + if (responseBody instanceof List) { + List<?> responseList = (List<?>) responseBody; + responseList.forEach(responseObserver::onNext); + } else { + responseObserver.onNext(responseBody); + } + return true; + } + + /** + * Sends the failure of the exchange to the client as an error, like for unary calls. + * + * @return true if the exchange failed and the error was sent + */ + protected boolean sendFailure(Exchange exchange) { + if (exchange.isFailed()) { + responseObserver.onError(GrpcMethodHandler.toStatusException(endpoint, exchange.getException())); Review Comment: 🟠**Potential NPE when `exchange.isFailed()` is true but `getException()` is null** `sendFailure()` checks `exchange.isFailed()` and passes `exchange.getException()` to `GrpcMethodHandler.toStatusException()`. In Camel, `isFailed()` can return `true` for fault messages where `getException()` returns `null`. `toStatusException()` then calls `cause.getMessage()` unconditionally when `muteException=false`, causing a `NullPointerException`. Add a null guard: ```suggestion if (exchange.isFailed()) { Exception ex = exchange.getException(); responseObserver.onError(GrpcMethodHandler.toStatusException(endpoint, ex != null ? ex : new IllegalStateException("Exchange failed with fault message"))); ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
