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 eee42db1f2b5 CAMEL-26550: camel-grpc - routeControlledStreamObserver
lets the route reply to streaming requests with the PROPAGATION strategy
(#29277)
eee42db1f2b5 is described below
commit eee42db1f2b50b850236b3e1ef7a26b4d8c16a4a
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Nov 6 08:28:21 2026 +0100
CAMEL-26550: camel-grpc - routeControlledStreamObserver lets the route
reply to streaming requests with the PROPAGATION strategy (#29277)
With routeControlledStreamObserver=true the PROPAGATION strategy lets the
route send the replies, and the consumer no longer sends its own, while each
onNext, onError and onCompleted is still handed to the route in order.
Co-Authored-By: Claude Opus 5.5 <[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 738d11cc4611..3e0e95612b3f 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
@@ -1190,6 +1190,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>
