Flow control support

[vietj]

No description provided.

[vietj]

@pmlopes I've started to implement flow-control for gRPC streams

I need your help on this feature as there is a timing issue when the GrpcWriteStreamImpl is created for a client that prevents adding a CallStreamObserver onReadyListener.

The failing test to look at is FlowControlTest#testStreamSink.

[vietj]

It looks to me that in the case of an GrpcUniExchange we create first the ReadStream and then the WriteStream and it looks to me that we should do both at the same time.

[vietj]

for ref https://github.com/vert-x3/vertx-grpc/tree/flow-control

[pmlopes]

The order of creation is a bit locked to the underlying API but I'll see if we can fix it.

[pmlopes]

@vietj the writeStream is initialized by gRPC itself in a private method:

    private static <ReqT, RespT> StreamObserver<ReqT> asyncStreamingRequestCall(ClientCall<ReqT, RespT> call, StreamObserver<RespT> responseObserver, boolean streamingResponse) {
        ClientCalls.CallToStreamObserverAdapter<ReqT> adapter = new ClientCalls.CallToStreamObserverAdapter(call);
        startCall(call, new ClientCalls.StreamObserverToCallListenerAdapter(call, responseObserver, adapter, streamingResponse), streamingResponse);
        return adapter;
    }

And then the class StreamObserverToCallListenerAdapter will call the freeze method on the observable, so when we try to set the setOnReadyHandler handler it will fail.

Also the transformation from pure StreamObserver to CallStreamObserver happens at the internal gRPC level (not at the code generation by the compiler) so there is no way to get a reference to that object before hand to set the handler.

[EmadAlblueshi]

Will be available in 3.5 ?

[vietj]

not an active area of development right now...

[anhldbk]

@vietj Do you need any help to resolve this issue?

[vietj]

any contribution would be valuable

[bbemis017]

Is there any update on this? or any known workarounds?