异步grpc存根中的取消订阅观察者-Java/静态编程语言

Question

我有一个异步存根，其中添加了一个观察者：

            val obs =  object: StreamObserver<Hallo> {

                override fun onNext(value: Hallo) {

                    streamSuccess(value)
                }

                override fun onError(t: Throwable?) {

                    nonSuccess(t?.message ?: "Unknow error")
                }

                override fun onCompleted() {

                    Log.d("Info", "completed")
                    completed()
                }
            }

我希望a能够从异步存根中删除这个观察者，这样我就可以取消客户端的流。

正如github上所说：https://github.com/grpc/grpc-java/issues/3095

我尝试保留观察者的一个局部变量，以便客户稍后可以执行以下操作：

observer?.onError(Status.CANCELLED.cause)

那不起作用。

我还尝试从抽象类创建自己的类：ClientCallStreamObserver

class CancellableStreamObserver<TResponse>(val next:(value:TResponse)->Unit, val onError:(t:Throwable)-> Unit, val onCompleted:(()->Unit), val onCanceledHandler: (()->Unit)? = null) : ClientCallStreamObserver<TResponse>() {
        override fun isReady(): Boolean {
            return  true
        }

        override fun setOnReadyHandler(onReadyHandler: Runnable?) {
            //TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
        }

        override fun disableAutoInboundFlowControl() {
            //TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
        }

        override fun cancel(message: String?, cause: Throwable?) {

            //TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
        }

        override fun request(count: Int) {
            //TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
        }

        override fun setMessageCompression(enable: Boolean) {
            //TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
        }

        override fun onNext(value: TResponse) {
            next(value)
        }

        override fun onError(t: Throwable) {
            if (t is StatusException) {
                if (t.status.code == Status.Code.CANCELLED) {
                    onCanceledHandler?.let {
                        it()
                    }
                }
            }
            if (t is StatusRuntimeException) {
                if (t.status.code == Status.Code.CANCELLED) {
                    onCanceledHandler?.let {
                        it()
                    }
                }
            }
            this.onError(t)
        }

        override fun onCompleted() {
            onCompleted()
        }
    }

所以稍后我可以打电话：

        observer?.cancel("Cancelled for the user",Status.CANCELLED.cause)

那也没用。

我知道它不起作用，因为如果用户再次添加新的观察者，我会得到重复的响应，就好像旧的观察者还活着一样。

我知道我可以用频道关闭频道。shutdownNow（）。但我觉得这太咄咄逼人了。

谢谢

Answer 1

参考https://github.com/grpc/grpc-java/issues/3095：

对于async，可以使用ClientCallStreamObserver。取消（），方法是将返回的StreamObserver强制转换为ClientCallStreamObserver，或实现传入的StreamObserver实现ClientResponseObserver。

（着重部分由作者标明）

grpc java将实现适当的方法，而不是您的实例。所以模式是：

stub.foo(req, object: ClientResponseObserver<Hallo> {
    override fun beforeStart(respObs: ClientCallStreamObserver<Hallo>) {
        // save respObs for later
    }
    override fun onNext(value: Hallo) {
        streamSuccess(value)
    }
    override fun onError(t: Throwable?) {
        nonSuccess(t?.message ?: "Unknow error")
    }
    override fun onCompleted() {
        Log.d("Info", "completed")
        completed()
    }
});

// -or- (for streaming calls only)

val obs = ...;
val respObs = stub.foo(obs) as (ClientCallStreamObserver<Hallo>);
respObs.onNext(req);
// save respObs for later

请注意，在这两种情况下，respObs是相同的。使用ClientResponseObserver主要是为了在响应观察者中进行流式处理并希望取消，以避免线程竞争。