gRPC 服务器如何注意到客户端取消了服务器端的流式调用？

Question

我想使用 gRPC 让客户端订阅服务器生成的事件。我有一个这样声明的 RPC：

rpc Subscribe (SubscribeRequest) returns (stream SubscribeResponse);

返回的流是无限的。要“取消订阅”，客户端会取消 RPC（顺便说一句。有更清洁的方法吗？）。

我已经弄清楚客户如何取消通话：

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

但是，服务器似乎没有注意到客户端已取消其调用。我正在使用虚拟服务器实现对此进行测试：

@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());
  }
}

服务器将愉快地继续将其消息发送到以太。服务器如何识别呼叫已被客户端取消，从而停止发送响应？

Answer 1

我自己找到了答案。您将传递的StreamObserver 转换为订阅ServerCallStreamObserver，它公开了方法isCancelled 和setOnCancelHandler。

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

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

对我来说，这引出了一个问题，为什么 subscribe 没有通过 ServerCallStreamObserver 开始。