Grpc client call

GrpcClient.kt

import com.google.protobuf.GeneratedMessageV3
import io.grpc.Channel
import io.grpc.ManagedChannel
import io.grpc.ManagedChannelBuilder
import io.grpc.stub.AbstractStub
import io.grpc.stub.StreamObserver
import java.util.concurrent.CompletableFuture

fun <S : AbstractStub<S>, I, R : GeneratedMessageV3> grpcCall(stub: S, call: S.(I, StreamObserver<R>) -> Unit): (I) -> CompletableFuture<R> {
    return { input ->
        val responseObserver = MyGrpcStreamObserver<R>()
        stub.call(input, responseObserver)
        responseObserver.result
    }
}

fun <S : AbstractStub<S>, I, R : GeneratedMessageV3> grpcCall(
        channelFactory: () -> Channel,
        stubFactory: (Channel) -> S,
        call: S.(I, StreamObserver<R>) -> Unit): (I) -> CompletableFuture<R> {
    return { input ->
        val responseObserver = MyGrpcStreamObserver<R>()
        stubFactory(channelFactory.invoke()).call(input, responseObserver)
        responseObserver.result
    }
}

public class MyGrpcStreamObserver<RESPONSE extends GeneratedMessageV3> implements StreamObserver<RESPONSE> {

    @Getter final CompletableFuture<RESPONSE> result = new CompletableFuture<RESPONSE>();

    @Override
    public void onNext(RESPONSE r) {
        result.complete(r);
    }

    @Override
    public void onError(Throwable t) {
        result.completeExceptionally(t);
    }

    @Override
    public void onCompleted() {
    }
}