This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26550-grpc-propagation-route-controlled in repository https://gitbox.apache.org/repos/asf/camel.git
commit a6ef31eda636fdb97cc6b0ead59b5ca2b368a8d1 Author: Claus Ibsen <[email protected]> AuthorDate: Thu Nov 5 20:42:19 2026 +0100 CAMEL-26550: camel-grpc - routeControlledStreamObserver lets the route reply to streaming requests with the PROPAGATION strategy With routeControlledStreamObserver=true and the default PROPAGATION consumer strategy, client streaming and bidirectional streaming calls ignored the option: the consumer sent the body at the end of the route as the reply and ended the call itself, and the route had no grpcResponseObserver, so a toRouteControlledStreamObserver producer failed. The PROPAGATION observer now sets the route-controlled exchange properties, with a failed flag shared by the events of the call, and leaves the replies and the end of the call to the route, as DELEGATION does. The events are still routed one at a time, in order. 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 | 6 +- .../org/apache/camel/component/grpc/grpc.json | 2 +- .../camel-grpc/src/main/docs/grpc-component.adoc | 6 +- .../camel/component/grpc/GrpcConfiguration.java | 4 +- .../GrpcRequestPropagationStreamObserver.java | 46 ++++-- .../grpc/GrpcPropagationRouteControlledTest.java | 178 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 ++ .../endpoint/dsl/GrpcEndpointBuilderFactory.java | 10 +- 9 files changed, 242 insertions(+), 23 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 b474fa62a1a1..271e0b643c90 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 @@ -49,7 +49,7 @@ "forwardOnError": { "index": 5, "kind": "parameter", "displayName": "Forward On Error", "group": "common", "label": "common", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "Determines if onError events should be pushed to the Camel route. Exceptions will be set a [...] "maxMessageSize": { "index": 6, "kind": "parameter", "displayName": "Max Message Size", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 4194304, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "The maximum message size allowed to be received\/sent (MiB)" }, "autoDiscoverServerInterceptors": { "index": 7, "kind": "parameter", "displayName": "Auto Discover Server Interceptors", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "Setting the autoDiscoverServerInterceptors mechanism, [...] - "consumerStrategy": { "index": 8, "kind": "parameter", "displayName": "Consumer Strategy", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "org.apache.camel.component.grpc.GrpcConsumerStrategy", "enum": [ "AGGREGATION", "PROPAGATION", "DELEGATION" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "PROPAGATION", "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configura [...] + "consumerStrategy": { "index": 8, "kind": "parameter", "displayName": "Consumer Strategy", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "org.apache.camel.component.grpc.GrpcConsumerStrategy", "enum": [ "AGGREGATION", "PROPAGATION", "DELEGATION" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "PROPAGATION", "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configura [...] "initialFlowControlWindow": { "index": 9, "kind": "parameter", "displayName": "Initial Flow Control Window", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1048576, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "Sets the initial flow control window in bytes." }, "keepAliveTime": { "index": 10, "kind": "parameter", "displayName": "Keep Alive Time", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 7200000, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "Sets a custom keepalive time in milliseconds, the delay time for sending next keepalive [...] "keepAliveTimeout": { "index": 11, "kind": "parameter", "displayName": "Keep Alive Timeout", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 20000, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "Sets a custom keepalive timeout in milliseconds, the timeout for keepalive ping requ [...] 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 918d1db67603..0387c2613597 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 @@ -98,7 +98,7 @@ exchange with an `IllegalArgumentException`; the call still ends with an `INTERN 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. +consumer sets this header for the events of client streaming and bidirectional streaming calls, 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 @@ -243,7 +243,9 @@ Therefore, you need the same technique as described in the server streaming sect But there is another thing: requests also come in streaming mode. So you need the following: -1. Set consumer strategy to DELEGATION - that differs from the default PROPAGATION option in the fact that consumer will not produce responses at all. If you set PROPAGATION, then you will receive more responses than you expected; +1. Choose the consumer strategy. With `routeControlledStreamObserver=true` the consumer sends no responses itself with either strategy: +the default `PROPAGATION` routes the events of a call one at a time, in order, while `DELEGATION` does not wait for an +event to be done before it routes the next one; 2. Forward `onError` and `onCompletion` on consumer; 3. Set producer strategy to STREAMING. 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 b474fa62a1a1..271e0b643c90 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 @@ -49,7 +49,7 @@ "forwardOnError": { "index": 5, "kind": "parameter", "displayName": "Forward On Error", "group": "common", "label": "common", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "Determines if onError events should be pushed to the Camel route. Exceptions will be set a [...] "maxMessageSize": { "index": 6, "kind": "parameter", "displayName": "Max Message Size", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 4194304, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "The maximum message size allowed to be received\/sent (MiB)" }, "autoDiscoverServerInterceptors": { "index": 7, "kind": "parameter", "displayName": "Auto Discover Server Interceptors", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "Setting the autoDiscoverServerInterceptors mechanism, [...] - "consumerStrategy": { "index": 8, "kind": "parameter", "displayName": "Consumer Strategy", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "org.apache.camel.component.grpc.GrpcConsumerStrategy", "enum": [ "AGGREGATION", "PROPAGATION", "DELEGATION" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "PROPAGATION", "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configura [...] + "consumerStrategy": { "index": 8, "kind": "parameter", "displayName": "Consumer Strategy", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "org.apache.camel.component.grpc.GrpcConsumerStrategy", "enum": [ "AGGREGATION", "PROPAGATION", "DELEGATION" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "PROPAGATION", "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configura [...] "initialFlowControlWindow": { "index": 9, "kind": "parameter", "displayName": "Initial Flow Control Window", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1048576, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "Sets the initial flow control window in bytes." }, "keepAliveTime": { "index": 10, "kind": "parameter", "displayName": "Keep Alive Time", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 7200000, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "Sets a custom keepalive time in milliseconds, the delay time for sending next keepalive [...] "keepAliveTimeout": { "index": 11, "kind": "parameter", "displayName": "Keep Alive Timeout", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 20000, "configurationClass": "org.apache.camel.component.grpc.GrpcConfiguration", "configurationField": "configuration", "description": "Sets a custom keepalive timeout in milliseconds, the timeout for keepalive ping requ [...] diff --git a/components/camel-grpc/src/main/docs/grpc-component.adoc b/components/camel-grpc/src/main/docs/grpc-component.adoc index 918d1db67603..0387c2613597 100644 --- a/components/camel-grpc/src/main/docs/grpc-component.adoc +++ b/components/camel-grpc/src/main/docs/grpc-component.adoc @@ -98,7 +98,7 @@ exchange with an `IllegalArgumentException`; the call still ends with an `INTERN 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. +consumer sets this header for the events of client streaming and bidirectional streaming calls, 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 @@ -243,7 +243,9 @@ Therefore, you need the same technique as described in the server streaming sect But there is another thing: requests also come in streaming mode. So you need the following: -1. Set consumer strategy to DELEGATION - that differs from the default PROPAGATION option in the fact that consumer will not produce responses at all. If you set PROPAGATION, then you will receive more responses than you expected; +1. Choose the consumer strategy. With `routeControlledStreamObserver=true` the consumer sends no responses itself with either strategy: +the default `PROPAGATION` routes the events of a call one at a time, in order, while `DELEGATION` does not wait for an +event to be done before it routes the next one; 2. Forward `onError` and `onCompletion` on consumer; 3. Set producer strategy to STREAMING. diff --git a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConfiguration.java b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConfiguration.java index b5154c051d44..a2a4a3de7304 100644 --- a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConfiguration.java +++ b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConfiguration.java @@ -349,7 +349,9 @@ public class GrpcConfiguration { * to the stream, and the response will be immediately sent back to the sender. If a delegation strategy is * selected, request is sent to the stream, but no response generated under the assumption that all necessary * responses will be sent at another part of route. Delegation strategy always comes with - * routeControlledStreamObserver=true to be able to achieve the assumption. + * routeControlledStreamObserver=true to be able to achieve the assumption. With routeControlledStreamObserver=true, + * the propagation strategy does not send responses either: it sends the requests to the route one at a time, and + * the route sends the responses. */ public GrpcConsumerStrategy getConsumerStrategy() { return consumerStrategy; 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 eaa49b450fa2..f43973357fe6 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 @@ -18,28 +18,37 @@ package org.apache.camel.component.grpc.server; import java.util.Map; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; import io.grpc.stub.StreamObserver; +import org.apache.camel.Exchange; import org.apache.camel.component.grpc.GrpcConsumer; import org.apache.camel.component.grpc.GrpcEndpoint; /** * This is the default consumer strategy for client-streaming and bi-directional streaming gRPC calls. gRPC request - * stream observer which is propagating every onNext(), onError() or onCompleted() calls to the Camel route. + * stream observer which is propagating every onNext(), onError() or onCompleted() calls to the Camel route, one event + * at a time. + * <p> + * With routeControlledStreamObserver=true the route sends the replies and ends the call itself, as with the DELEGATION + * strategy: the consumer sends nothing to the client. */ public class GrpcRequestPropagationStreamObserver extends GrpcRequestAbstractStreamObserver { + private final boolean routeControlled; + // set once the call ended with an error, by a failed exchange or by the route: nothing more can be sent to the + // client. Shared by the exchanges of all the events of the call when the route controls the replies + private final AtomicBoolean failed = new AtomicBoolean(); + public GrpcRequestPropagationStreamObserver(GrpcEndpoint endpoint, GrpcConsumer consumer, StreamObserver<Object> responseObserver, Map<String, Object> headers) { super(endpoint, consumer, responseObserver, headers); + this.routeControlled = endpoint.getConfiguration().isRouteControlledStreamObserver(); } - // set when a failed exchange ended the call with an error: nothing more can be sent to the client - private volatile boolean failed; - @Override public void onNext(Object request) { - if (failed) { + if (failed.get()) { return; } CountDownLatch latch = new CountDownLatch(1); @@ -47,24 +56,25 @@ public class GrpcRequestPropagationStreamObserver extends GrpcRequestAbstractStr exchange = endpoint.createExchange(); exchange.getIn().setBody(request); exchange.getIn().setHeaders(headers); + setRouteControlledProperties(exchange); consumer.process(exchange, doneSync -> latch.countDown()); try { latch.await(); - if (sendFailure(exchange)) { - failed = true; + if (routeControlled) { + // the route sends the replies itself return; } - - if (!sendResponses(exchange)) { - failed = true; + if (sendFailure(exchange) || !sendResponses(exchange)) { + failed.set(true); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); - responseObserver.onError(e); - failed = true; + if (failed.compareAndSet(false, true)) { + responseObserver.onError(e); + } } } @@ -72,8 +82,9 @@ public class GrpcRequestPropagationStreamObserver extends GrpcRequestAbstractStr public void onError(Throwable throwable) { exchange = endpoint.createExchange(); exchange.getIn().setHeaders(headers); + setRouteControlledProperties(exchange); consumer.onError(exchange, throwable); - if (!failed) { + if (!routeControlled && !failed.get()) { responseObserver.onError(throwable); } } @@ -82,9 +93,16 @@ public class GrpcRequestPropagationStreamObserver extends GrpcRequestAbstractStr public void onCompleted() { exchange = endpoint.createExchange(); exchange.getIn().setHeaders(headers); + setRouteControlledProperties(exchange); consumer.onCompleted(exchange); - if (!failed) { + if (!routeControlled && !failed.get()) { responseObserver.onCompleted(); } } + + private void setRouteControlledProperties(Exchange exchange) { + if (routeControlled) { + GrpcMethodHandler.setRouteControlledProperties(endpoint, exchange, responseObserver, responseType, failed); + } + } } diff --git a/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcPropagationRouteControlledTest.java b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcPropagationRouteControlledTest.java new file mode 100644 index 000000000000..de700ae8b8b6 --- /dev/null +++ b/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; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.spi.CamelEvent.ExchangeEvent; +import org.apache.camel.support.EventNotifierSupport; +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.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * With the default PROPAGATION consumer strategy and routeControlledStreamObserver=true, the route sends the replies of + * client streaming and bidirectional streaming calls with the toRouteControlledStreamObserver producer: the consumer + * sends nothing itself. + */ +public class GrpcPropagationRouteControlledTest extends GrpcTestSupport { + + private static final String CONSUMER = "grpc://localhost:0/org.apache.camel.component.grpc.PingPong?synchronous=true" + + "&consumerStrategy=PROPAGATION&routeControlledStreamObserver=true" + + "&forwardOnCompleted=true&muteException=false"; + 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<>(); + private final AtomicInteger clientStreamingPings = new AtomicInteger(); + + @AfterEach + public void stopGrpcChannels() { + channels.values().forEach(channel -> channel.shutdown().shutdownNow()); + channels.clear(); + } + + @Test + public void bidiStreamingRouteSendsOneReplyPerEvent() throws Exception { + PongResponseStreamObserver responseObserver = new PongResponseStreamObserver(); + StreamObserver<PingRequest> requestObserver + = PingPongGrpc.newStub(channel("propagation-bidi")).pingAsyncAsync(responseObserver); + requestObserver.onNext(pingRequest("PING1")); + requestObserver.onNext(pingRequest("PING2")); + requestObserver.onCompleted(); + assertTrue(responseObserver.latch.await(5, TimeUnit.SECONDS)); + + assertNull(responseObserver.error, "the call must not end with an error"); + assertEquals(2, responseObserver.pongResponses.size(), "exactly one reply per event: " + + responseObserver.pongResponses); + assertEquals("PING1PONG", responseObserver.pongResponses.get(0).getPongName()); + assertEquals("PING2PONG", responseObserver.pongResponses.get(1).getPongName()); + assertEquals(1, responseObserver.completed.get()); + // two onNext exchanges and the forwarded onCompleted exchange, none failed + assertEquals(Collections.nCopies(3, null), awaitDoneExchanges("propagation-bidi", 3)); + } + + @Test + public void clientStreamingRouteSendsTheReplyOnCompleted() throws Exception { + PongResponseStreamObserver responseObserver = new PongResponseStreamObserver(); + StreamObserver<PingRequest> requestObserver + = PingPongGrpc.newStub(channel("propagation-client-streaming")).pingAsyncSync(responseObserver); + requestObserver.onNext(pingRequest("PING1")); + requestObserver.onNext(pingRequest("PING2")); + requestObserver.onCompleted(); + assertTrue(responseObserver.latch.await(5, TimeUnit.SECONDS)); + + // the consumer does not send the unchanged requests of the onNext events as replies + assertNull(responseObserver.error, "the call must not end with an error"); + assertEquals(1, responseObserver.pongResponses.size(), "exactly one reply: " + responseObserver.pongResponses); + assertEquals("PONG2", responseObserver.pongResponses.get(0).getPongName()); + assertEquals(1, responseObserver.completed.get()); + assertEquals(Collections.nCopies(3, null), awaitDoneExchanges("propagation-client-streaming", 3)); + } + + 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 PingRequest pingRequest(String name) { + return PingRequest.newBuilder().setPingName(name).setPingId(1).build(); + } + + private ManagedChannel channel(String routeId) { + return channels.computeIfAbsent(routeId, + id -> ManagedChannelBuilder.forAddress("localhost", getRoutePort(id)).usePlaintext().build()); + } + + @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; + } + }); + + // bidirectional streaming: a reply for each request, then the forwarded onCompleted ends the call + from(CONSUMER + "&hash=1") + .routeId("propagation-bidi") + .process(e -> { + PingRequest request = e.getMessage().getBody(PingRequest.class); + if (request != null) { + e.getMessage().setBody(PongResponse.newBuilder() + .setPongName(request.getPingName() + "PONG").setPongId(request.getPingId()) + .build()); + } + }) + .to(REPLY); + + // client streaming: the requests are counted, and the single reply is sent on the forwarded onCompleted + from(CONSUMER + "&hash=2") + .routeId("propagation-client-streaming") + .choice() + .when(header(GrpcConstants.GRPC_EVENT_TYPE_HEADER).isEqualTo(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT)) + .process(e -> clientStreamingPings.incrementAndGet()) + .otherwise() + .process(e -> e.getMessage().setBody( + PongResponse.newBuilder().setPongName("PONG" + clientStreamingPings.get()).build())) + .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) + .end(); + } + }; + } +} 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 8d76abb015b2..4b4d552c24d2 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 @@ -1137,6 +1137,17 @@ exchange, where it sent nothing before. An `onError` event with a body that is n 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-grpc - routeControlledStreamObserver applies to streaming requests with the PROPAGATION strategy + +With `routeControlledStreamObserver=true` and the default `PROPAGATION` consumer strategy, the route now sends the +replies of client streaming and bidirectional streaming calls, as it already did for unary and server streaming calls. +The consumer passes the `grpcResponseObserver` exchange property to the route, and no longer sends the body at the end +of the route or ends the call itself. Before, a producer with `toRouteControlledStreamObserver=true` failed with an +`IllegalArgumentException`, and the consumer sent the body at the end of the route as the reply. Such a route must now +send its replies with a producer with `toRouteControlledStreamObserver=true` and end the call, for example on the event +forwarded with `forwardOnCompleted=true`. The events of a call are still routed one at a time, in order, which is the +difference with the `DELEGATION` strategy. + === 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 diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GrpcEndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GrpcEndpointBuilderFactory.java index 32cb1e241016..8be192e47867 100644 --- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GrpcEndpointBuilderFactory.java +++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GrpcEndpointBuilderFactory.java @@ -210,7 +210,10 @@ public interface GrpcEndpointBuilderFactory { * stream, but no response generated under the assumption that all * necessary responses will be sent at another part of route. Delegation * strategy always comes with routeControlledStreamObserver=true to be - * able to achieve the assumption. + * able to achieve the assumption. With + * routeControlledStreamObserver=true, the propagation strategy does not + * send responses either: it sends the requests to the route one at a + * time, and the route sends the responses. * * The option is a: * <code>org.apache.camel.component.grpc.GrpcConsumerStrategy</code> @@ -237,7 +240,10 @@ public interface GrpcEndpointBuilderFactory { * stream, but no response generated under the assumption that all * necessary responses will be sent at another part of route. Delegation * strategy always comes with routeControlledStreamObserver=true to be - * able to achieve the assumption. + * able to achieve the assumption. With + * routeControlledStreamObserver=true, the propagation strategy does not + * send responses either: it sends the requests to the route one at a + * time, and the route sends the responses. * * The option will be converted to a * <code>org.apache.camel.component.grpc.GrpcConsumerStrategy</code>
