kotlin grpc.StreamObserver去除rx.PublishSubject

Question

每当我们声明流式api时使用GRPC

rpc heartBeat(Empty) returns (stream ServiceStatus){}

我们谷歌搜索观察者模式的简单接口StreamObserver(这是protobuf将为我们生成的)

public interface StreamObserver<V> {
  void onNext(V var1);

  void onError(Throwable var1);

  void onCompleted();
}

现在你想要做的是将它转换为实际,Observable并且只有在传递之后才能进一步使用.

override fun heartBeat(arg: Empty): Observable<ServiceStatus> {
    // we create rx java subject
    val subject = PublishSubject.create<ServiceStatus>()

    // we create grpc observer and delegate all calls to rx java
    val observer = object : StreamObserver<ServiceStatus> {

        override fun onNext(value: ServiceStatus) {
            subject.onNext(value)
        }

        override fun onError(error: Throwable) {
            subject.onError(error)
        }

        override fun onCompleted() {
            subject.onCompleted()
        }
    }

    // we use grpc observer for generated api
    asyncStub.heartBeat(arg, observer)

    // but we pass rx observable (subject) to client code
    return subject
}

现在我是Kotlin的新手,但我无法弄清楚现有的授权功能是否有办法让StreamObserver的主题委托？ 在Kotlin中编写这段代码是否有更具表现力的方法？

Answer 1

我将创建一个创建a的泛型方法StreamObserver,将其传递给它的lambda参数并将结果包装起来Observable.

inline fun <T> asObservable(
    crossinline body: (StreamObserver<T>) -> Unit): Observable<T> {
  return Observable.create { subscription ->
    val observer = object : StreamObserver<T> {
      override fun onNext(value: T) {
        subscription.onNext(value)
      }

      override fun onError(error: Throwable) {
        subscription.onError(error)
      }

      override fun onCompleted() {
        subscription.onCompleted()
      }
    }
    body(observer)
  }
}

然后,您可以通过以下方式实现RPC方法.

override fun heartBeat(arg: Empty): Observable<ServiceStatus> =
    asObservable { asyncStub.heartBeat(arg, it) }