How can a gRPC server notice that the client has cancelled a server-side streaming call?

Question

I want to use gRPC to let clients subscribe to events generated by the server. I have an RPC declared like so:

rpc Subscribe (SubscribeRequest) returns (stream SubscribeResponse);

where the returned stream is infinite. To "unsubscribe", clients cancel the RPC (btw. is there a cleaner way?).

I have figured out how the client can cancel the call:

Context.CancellableContext cancellableContext =
         Context.current().withCancellation();
cancellableContext.run(() -> {
   stub.subscribe(request, callback);
});
// do other stuff / wait for reason to unsubscribe
cancellableContext.cancel(new InterruptedException());

However, the server does not seem to notice that a client has cancelled its call. I'm testing this with a dummy server implementation:

@Override
public void subscribe(SubscribeRequest request,
                      StreamObserver<SubscribeResponse> responseObserver) {
  // in real code, this will happen in a separate thread.
  while (!Thread.interrupted()) {
    responseObserver.onNext(SubscribeResponse.getDefaultInstance());
  }
}

The server will happily continue sending its messages into the ether. How can the server recognize that the call was cancelled by the client and thus stop sending responses?

Balz Guenat Balz Guenat · Accepted Answer · 2019-02-08T09:38:15

I found the answer myself. You cast the StreamObserver passed to subscribe to a ServerCallStreamObserver, which exposes methods isCancelled and setOnCancelHandler.

scso = ((ServerCallStreamObserver<SubscribeResponse>) responseObserver);

scso.setOnCancelHandler(handler);
// or
if (scso.isCancelled()) {
  // do whatever
}

This, to me, begs the question why subscribe isn't passed a ServerCallStreamObserver to begin with.

How can a gRPC server notice that the client has cancelled a server-side streaming call?

1 Answers