This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 57387dfa7742 [backport camel-4.22.x] CAMEL-26550: camel-grpc -
routeControlledStreamObserver lets the route reply to streaming requests with
the PROPAGATION strategy (#29300)
57387dfa7742 is described below
commit 57387dfa774260a207085d80c2e001f3609c435c
Author: Guillaume Nodet <[email protected]>
AuthorDate: Mon Nov 9 13:01:25 2026 +0100
[backport camel-4.22.x] CAMEL-26550: camel-grpc -
routeControlledStreamObserver lets the route reply to streaming requests with
the PROPAGATION strategy (#29300)
* [backport camel-4.22.x] CAMEL-26550: camel-grpc -
routeControlledStreamObserver lets the route reply to streaming requests with
the PROPAGATION strategy
* fix: backport companion changes to GrpcRequestAbstractStreamObserver and
GrpcMethodHandler
The backport cherry-picked GrpcRequestPropagationStreamObserver.java which
references sendFailure(), sendResponses(), and responseType from the
abstract
base class, and setRouteControlledProperties()/toStatusException() from
GrpcMethodHandler — none of which exist in camel-4.22.x.
Add the minimum required:
- GrpcConstants: GRPC_RESPONSE_TYPE, GRPC_MUTE_EXCEPTION,
GRPC_RESPONSE_OBSERVER_FAILED
- GrpcRequestAbstractStreamObserver: responseType field, sendFailure(),
sendResponses()
- GrpcMethodHandler: setRouteControlledProperties(), toStatusException()
static methods
* fix: remove unused import java.util.List from
GrpcRequestPropagationStreamObserver
* fix: address review comments on backport PR #29300
- GrpcPropagationRouteControlledTest: inline PongResponseStreamObserver
as a static inner class instead of importing it from
GrpcConsumerReplyTypeTest,
which does not exist in the camel-4.22.x branch (it was introduced by
CAMEL-26187 / PR #29012)
- grpc-component.adoc: replace the type-conversion consumer-reply section
(copied from CAMEL-26187's conflict resolution) with an accurate
description
of what GrpcProducerToRouteControlledStreamObserver actually does in this
branch: forward onNext/onCompleted/onError events by CamelGrpcEventType
header, without type conversion
- GrpcMethodHandler.toStatusException(): add null guard on
cause.getMessage()
to prevent NPE when exchange.isFailed() is true but getException() returns
null (fault messages)
* fix: update @return Javadoc in sendResponses() to match simplified
backport behavior
* fix: address review — restore 4.22.x behavior in non-routeControlled
path, remove dead API
- Restore the original 4.22.x reply logic (body forwarded via onNext
unconditionally) in
the non-routeControlled branch of
GrpcRequestPropagationStreamObserver.onNext().
sendFailure() was introduced from main (CAMEL-25250 / #27247) which is
not on 4.22.x;
removing it avoids a behavioral regression for existing PROPAGATION users.
- Remove the dead sendFailure()/sendResponses() helpers and
responseType/getResponseType/
setResponseType from GrpcRequestAbstractStreamObserver (no callers on
this branch).
- Remove the static setRouteControlledProperties() and toStatusException()
methods from
GrpcMethodHandler (no callers after the above cleanup).
- Remove the three dead public API constants GRPC_RESPONSE_TYPE,
GRPC_MUTE_EXCEPTION and
GRPC_RESPONSE_OBSERVER_FAILED from GrpcConstants (unused on 4.22.x).
- Add an upgrade-guide entry for CAMEL-26550 in the 4.22.2→4.22.3 section.
* fix: remove unused Exchange import and fix import ordering in
GrpcRequestPropagationStreamObserver
* nit: fix comment on 'failed' field to reflect it is set in both
routeControlled and non-routeControlled modes
---
.../org/apache/camel/catalog/components/grpc.json | 2 +-
.../apache/camel/catalog/docs/grpc-component.adoc | 15 +-
.../org/apache/camel/component/grpc/grpc.json | 2 +-
.../camel-grpc/src/main/docs/grpc-component.adoc | 15 +-
.../camel/component/grpc/GrpcConfiguration.java | 4 +-
.../GrpcRequestPropagationStreamObserver.java | 43 ++++-
.../grpc/GrpcPropagationRouteControlledTest.java | 203 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 10 +
.../endpoint/dsl/GrpcEndpointBuilderFactory.java | 10 +-
9 files changed, 292 insertions(+), 12 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 05519fe07a38..09bb31224c06 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
@@ -43,7 +43,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 c092f6a8e8ed..b480d6c1b73a 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,17 @@ 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.
+
+With `routeControlledStreamObserver=true` the route sends the responses
itself, with a producer endpoint with
+`toRouteControlledStreamObserver=true`. The producer sends the event named by
the `CamelGrpcEventType` header
+(`onNext`, `onCompleted` or `onError`). The consumer sets this header for the
events of client streaming and
+bidirectional streaming calls. The body of an `onNext` event is sent to the
client as is. The body of an `onError`
+event must be a `Throwable`, which ends the call.
+
=== 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.
@@ -212,7 +223,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 05519fe07a38..09bb31224c06 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
@@ -43,7 +43,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 c092f6a8e8ed..b480d6c1b73a 100644
--- a/components/camel-grpc/src/main/docs/grpc-component.adoc
+++ b/components/camel-grpc/src/main/docs/grpc-component.adoc
@@ -73,6 +73,17 @@ 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.
+
+With `routeControlledStreamObserver=true` the route sends the responses
itself, with a producer endpoint with
+`toRouteControlledStreamObserver=true`. The producer sends the event named by
the `CamelGrpcEventType` header
+(`onNext`, `onCompleted` or `onError`). The consumer sets this header for the
events of client streaming and
+bidirectional streaming calls. The body of an `onNext` event is sent to the
client as is. The body of an `onError`
+event must be a `Throwable`, which ends the call.
+
=== 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.
@@ -212,7 +223,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 4764cceac0b7..514673e81a3a 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
@@ -19,35 +19,57 @@ package org.apache.camel.component.grpc.server;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
import io.grpc.stub.StreamObserver;
+import org.apache.camel.component.grpc.GrpcConstants;
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 due to an interrupt (both
routeControlled and non-routeControlled):
+ // nothing more can be sent to the client
+ 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();
}
@Override
public void onNext(Object request) {
+ if (routeControlled && failed.get()) {
+ return;
+ }
CountDownLatch latch = new CountDownLatch(1);
exchange = endpoint.createExchange();
exchange.getIn().setBody(request);
exchange.getIn().setHeaders(headers);
+ if (routeControlled) {
+ exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+ }
consumer.process(exchange, doneSync -> latch.countDown());
try {
latch.await();
+ if (routeControlled) {
+ // the route sends the replies itself
+ return;
+ }
Object responseBody = exchange.getMessage().getBody();
if (responseBody instanceof List) {
List<?> responseList = (List<?>) responseBody;
@@ -57,8 +79,9 @@ public class GrpcRequestPropagationStreamObserver extends
GrpcRequestAbstractStr
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- responseObserver.onError(e);
-
+ if (failed.compareAndSet(false, true)) {
+ responseObserver.onError(e);
+ }
}
}
@@ -66,15 +89,25 @@ public class GrpcRequestPropagationStreamObserver extends
GrpcRequestAbstractStr
public void onError(Throwable throwable) {
exchange = endpoint.createExchange();
exchange.getIn().setHeaders(headers);
+ if (routeControlled) {
+ exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+ }
consumer.onError(exchange, throwable);
- responseObserver.onError(throwable);
+ if (!routeControlled && !failed.get()) {
+ responseObserver.onError(throwable);
+ }
}
@Override
public void onCompleted() {
exchange = endpoint.createExchange();
exchange.getIn().setHeaders(headers);
+ if (routeControlled) {
+ exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER,
responseObserver);
+ }
consumer.onCompleted(exchange);
- responseObserver.onCompleted();
+ if (!routeControlled && !failed.get()) {
+ responseObserver.onCompleted();
+ }
}
}
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..12bb5ea8f384
--- /dev/null
+++
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcPropagationRouteControlledTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.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.stub.StreamObserver;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+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();
+ }
+ };
+ }
+
+ 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_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index de1af9af0608..f32382d9b5b2 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -214,6 +214,16 @@ nodes, and XInclude is not processed. Before, the Flowable
7 deserializer parsed
An event that relies on an external DTD or an external entity no longer gets
their content. A channel that sets
`deserializerDelegateExpression` keeps its own deserializer, which can parse
such events.
+=== camel-grpc - routeControlledStreamObserver now supported for the
PROPAGATION strategy
+
+The `routeControlledStreamObserver` option now works with the PROPAGATION
consumer strategy
+(CAMEL-26550). When enabled, the consumer dispatches each streaming event to
the route and the
+route is responsible for sending replies via
`toRouteControlledStreamObserver`, exactly as with
+the DELEGATION strategy.
+
+No changes are required for existing routes that use PROPAGATION without
+`routeControlledStreamObserver=true`.
+
=== camel-jsch - the scp producer refuses a file name with a line feed or a
NUL character
The `scp` producer now refuses a file name with a line feed or a NUL
character, in the file name or in one of its
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>
