gnodet-bot commented on code in PR #29300:
URL: https://github.com/apache/camel/pull/29300#discussion_r4511070479
##########
components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestPropagationStreamObserver.java:
##########
@@ -16,65 +16,93 @@
*/
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.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();
}
@Override
public void onNext(Object request) {
+ if (failed.get()) {
+ return;
+ }
CountDownLatch latch = new CountDownLatch(1);
exchange = endpoint.createExchange();
exchange.getIn().setBody(request);
exchange.getIn().setHeaders(headers);
+ setRouteControlledProperties(exchange);
consumer.process(exchange, doneSync -> latch.countDown());
try {
latch.await();
- Object responseBody = exchange.getMessage().getBody();
- if (responseBody instanceof List) {
- List<?> responseList = (List<?>) responseBody;
- responseList.forEach(responseObserver::onNext);
- } else {
- responseObserver.onNext(responseBody);
+ if (routeControlled) {
+ // the route sends the replies itself
+ return;
+ }
+ if (sendFailure(exchange) || !sendResponses(exchange)) {
Review Comment:
**Behavioral regression — non-routeControlled PROPAGATION path.**
On 4.22.x, `sendFailure()` is new in this branch. When
`routeControlled=false` and the exchange fails:
- Old behaviour: `responseObserver.onNext(body)` was called (body forwarded
even on failure).
- New behaviour: `responseObserver.onError(INTERNAL)` is sent, and `failed`
is set to `true`, which silently drops all subsequent events for the call.
If this bug fix (CAMEL-25250 / original PR #29277) is intentional here,
please:
1. Add an upgrade-guide entry on `main` documenting the behaviour change.
2. Note it in the PR description so backport reviewers can make an informed
decision.
If it was introduced accidentally during conflict resolution, remove
`sendFailure(exchange)` from this branch and keep only
`sendResponses(exchange)`.
Also: `!sendResponses(exchange)` is dead code — that method always returns
`true`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
