java - kotlin grpc.StreamObserver删除为rx.PublishSubject

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

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

我们有用于观察者模式StreamObserver的googles简单界面（这是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的Subject委托？
有没有一种更富有表现力的方式可以在Kotlin中编写这段代码？

最佳答案

我将创建一个创建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) }