How to cancel gRPC call from client side?

[AlekseyMay]

There is a 3rd party server which is streaming to my quarkus app (3.2.0.Final) with oneToMany (input -> output Multi). And I need to close the connection on certain condition (by filtering Object status of last event). Otherwise the server can't understand that the stream must be closed after the last event it sent to the client and right now the only way I ve found is to use deadLine to close such connection.
I ve found that it is possible to do so with such method call: grpc/grpc-java#3115

I'm using Mutiny gRPC generation in my app and I can't understand how to implement it using it.

Any help appreciated.

[quarkus-bot[bot]]

/cc @alesj (grpc), @cescoffier (grpc)

[cescoffier]

I would need more details about your code.

I can see two ways:

If the client sends a stream, sending the completion signal should close the connection

If you consume a stream from the server, you have a subscription on it. Calling cancel on that subscription will close the connection.

[AlekseyMay]

Thank you for the reply.

I consume a stream on client side from 3rd party server. And I have custom Mutiny operator for this case which filters events up to DONE status. And use onComplete() on DONE event. Maybe I ve done smth wrong or it is not the correct way? I ve tried your suggestion to call subscription.cancel() on DONE status, but nothing changed. My class looks like this:

@Log4j2
public class NotificationOp extends AbstractMultiOperator<WatchOperationsResponse, Operation> {

    public NotificationOp(Multi<WatchOperationsResponse> upstream) {
        super(upstream);
    }

    @Override
    public void subscribe(MultiSubscriber<? super Operation> downstream) {
        super.upstream.subscribe().withSubscriber(new AccumulatorProcessor(downstream));
    }

    private static class AccumulatorProcessor extends MultiOperatorProcessor<WatchOperationsResponse, Operation> {

        private AccumulatorProcessor(MultiSubscriber<? super Operation> downstream) {
            super(downstream);
        }

        @Override
        public void onItem(WatchOperationsResponse resp) {
            try {
                if (super.isDone())
                    return;
                if (resp.hasInitialState()) {
                    final Operation operation = resp.getInitialState().getOperations(0);
                    log.info("Initial status value of operation with id: {} and group: {} => status: {}", operation.getId(), operation.getGroup(), operation.getState());
                    filterStatusAndComplete(operation);
                } else if (resp.hasInitFinished()) {
                    log.info("Initial status was sent, waiting operation to reach status {}", OperationState.DONE);
                    super.request(1);
                } else if (resp.hasUpdate()) {
                    final Operation operation = resp.getUpdate().getOperations(0);
                    log.info("Operation with id: {} and group: {} changed it's status => status: {}", operation.getId(), operation.getGroup(), operation.getState());
                    filterStatusAndComplete(operation);
                }
            } catch (Exception ex) {
                failAndCancel(ex);
            }
        }

        private void filterStatusAndComplete(Operation operation) {
            if (OperationState.DONE.equals(operation.getState())) {
                log.info("Operation with id: {} was finished", operation.getId());
                this.downstream.onItem(operation);
                this.downstream.onCompletion();
            } else if (OperationState.FAILED.equals(operation.getState()) ||
                       OperationState.UNRECOGNIZED.equals(operation.getState())) {
                log.error("Error happened for operation with id: {}, status: {}", operation.getId(), operation.getState());
                failAndCancel(new IllegalStateException("Failed to create script on operation with id: %s, status: %s".formatted(operation.getId(), operation.getState())));
            } else {
                super.request(1);
            }
        }

        private void failAndCancel(Exception ex) {
            final Flow.Subscription subscription = super.getUpstreamSubscription();
            if (subscription != null) {
                subscription.cancel();
            }
            super.onFailure(ex);
        }

    }

}

Another strange thing here is that I have simular case between my microservices, written on Quarkus, when client consumes stream from server and everything is ok.

[AlekseyMay]

@cescoffier
Finally, I ve found the way that works correctly and cancels the stream from client side. But this solution is not great (:
I filter in the interceptor for special status and cancel call inside interceptor, where I have access to ClientCall.

@ApplicationScoped
public class GrpcInterceptor implements ClientInterceptor {

    private static final Metadata.Key<String> AUTHORIZATION = Metadata.Key.of(HttpHeaders.AUTHORIZATION, Metadata.ASCII_STRING_MARSHALLER);

    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,
                                                               CallOptions callOptions, Channel next) {

        return new ForwardingClientCall.SimpleForwardingClientCall<>(
                next.newCall(method, callOptions)
        ) {
        
            private void cancelCall() {
                this.cancel(Status.Code.CANCELLED.name(), null);
                
            }
            @Override
            public void start(Listener<RespT> responseListener, Metadata headers) {
                headers.put(AUTHORIZATION, AuthGeneration.createAuthHeader(headers.get(SERVICE_ENDPOINT)));
                headers.removeAll(SERVICE_ENDPOINT);

                super.start(new ForwardingClientCallListener.SimpleForwardingClientCallListener<>(responseListener) {

                    @Override
                    public void onMessage(RespT message) {
                        super.onMessage(message);
                        if (message instanceof WatchOperationsResponse response
                                && response.hasUpdate()
                                && OperationState.DONE.equals(response.getUpdate().getOperations(0).getState())) {
                            cancelCall();
                        }
                    }

                    @Override
                    public void onClose(Status status, Metadata trailers) {
                        if (Status.CANCELLED.getCode().equals(status.getCode())) {
                            super.onClose(Status.OK, trailers);
                        } else {
                            super.onClose(status, trailers);
                        }
                    }

                }, headers);
            }

        };
    }

}

I am sure that better solution exists. I ve tried cancelling upstream subscription inside custom Mutiny operator, but unfortunately that changed nothing and the stream continues working until deadLine activates and kill it. And again the strange problem exists only between 3rd party server and Quarkus client app. If it is Quarkus server and Quarkus client consuming Multi, all is ok on client side, without any additional cancellation.

[cescoffier]

It might be possible that the service does not handle the cancellation correctly. Unfortunately, if that's the case, I don't think there will be any good solution.
You can try to run Wireshark and check the frames.

[jmini]

Related to this discussion, following issues looks interesting:

• Mutiny subscription is not cancelled when grpc client is cancelled #41776
• gRPC client closing the stream doesn't automatically propagate the cancellation signal #46986

With this PR potentially fixing #47555 it.
But on hold because of this issue in Vert.x: eclipse-vertx/vert.x#5562