Implement outbound flow control

[ejona86]

Should be token-based a la reactive streams. Although it also seems that maybe we will only ever have at most 1 token passed to the application.

Currently we provide no method of pushback to the application and buffer infinitely as the application sends.

[nmittler]

@ejona86 @louiscryan I'm looking at the right place add this...

I suppose I can just add a request(numMessages) method to Call.Listener and then have Call.sendPayload throw if no message has been requested. Thoughts?

[ejona86]

That requires StreamObserver to buffer, unless we expose flow control to the applications. Would it be better to be advisory and code is allowed to ignore it if it doesn't care about flow control? Or does this encourage us to adopt full reactive-streams for StreamObserver and require applications to abide by flow control?

[nmittler]

I was thinking that the stub layer would block until notified since we don't expose flow control at the stub layer. Is that not the behavior we had discussed in the past?

[ejona86]

I still feel like we can't block when the application calls StreamObserver.onValue. If it can block, then what is the point of going through the agony of using an async interface? If it can block, I suggest we just trash it altogether and tell people to use the blocking interface.

[nmittler]

Ah right, good point. So I guess we should expose outbound flow control for async clients?

[ejona86]

@louiscryan What are your ideas for outbound flow control?

[nmittler]

Discussed offline. Here's the idea...

Basically we just extend Call.Listener to have the reactivestreams-style request method.

interface Call.Listener {
    void onHeaders(...);
    void onPayload(...);
    void onClose(...);
    void onReady(int numMessages);
}

For now I think the method should be called onReady rather than request. First, request is more of a command which seems out of place in a listener interface. Second, I don't think we'll want to initially be as rigid as the reactivestreams API and fail if the user tries to send more data than has been request. Instead we'll let the user opt-in to outbound flow control, similar to stubby. This will allow the async stub to simply send data immediately without requiring any additional buffering.

I originally considered that onReady would have no arguments, however if we tell the listener the number of messages that it may send, this gives us some flexibility if we find that requesting one message at a time causes performance hiccups.

WDYT?

[nmittler]

@louiscryan @ejona86 any thoughts before I start hacking? :)

[ejona86]

@nmittler, SGTM.

[JakeWharton]

Request is a command in reactive streams hence its lack of "on". It's the
onStart or onNext callbacks issuing a command for data back upwards to the
source.
On Feb 12, 2015 10:36 AM, "Nathan Mittler" notifications@github.com wrote:

Discussed offline. Here's the idea...

Basically we just extend Call.Listener to have the reactivestreams-style
request method.

interface Call.Listener {
void onHeaders(...);
void onPayload(...);
void onClose(...);
void onReady(int numMessages);
}

For now I think the method should be called onReady rather than request.
First, request is more of a command which seems out of place in a
listener interface. Second, I don't think we'll want to initially be as
rigid as the reactivestreams API and fail if the user tries to send more
data than has been request. Instead we'll let the user opt-in to outbound
flow control, similar to stubby. This will allow the async stub to simply
send data immediately without requiring any additional buffering.

I originally considered that onReady would have no arguments, however if
we tell the listener the number of messages that it may send, this gives us
some flexibility if we find that requesting one message at a time causes
performance hiccups.

WDYT?

—
Reply to this email directly or view it on GitHub
#6 (comment).

[nmittler]

Fixed by #348