Skip to content

Commit 764a857

Browse files
committed
stub: Clarify flow control javadoc
Remove vestigial references to credit-based flow control and the suggestion that a request() for messages directly corresponds to the number of messages a peer could subsequently send before its end of the stream went !isReady(). In fact, okhttp, netty and binder transports all use buffers meaning the sender's stream can be isReady() even when the receiver has no outstanding request()s for messages. Furthermore, those buffers are sized in bytes and messages are not all the same size. So consuming an inbound's next message may not cause a non-ready outbound to become ready again. Replace this with a short discussion of what is actually guaranteed by every transport.
1 parent 6b2d978 commit 764a857

3 files changed

Lines changed: 24 additions & 9 deletions

File tree

stub/src/main/java/io/grpc/stub/CallStreamObserver.java

Lines changed: 22 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,23 @@
4747
* <p>Like {@code StreamObserver}, implementations are not required to be thread-safe; if multiple
4848
* threads will be writing to an instance concurrently, the application must synchronize its calls.
4949
*
50+
* <p>On flow control: The {@link #isReady} state of an outbound {@link CallStreamObserver} is
51+
* related to the number and size of messages written to its {@link StreamObserver#onNext} and
52+
* whether the peer has consumed those messages by way of {@link StreamObserver#onNext} on the
53+
* corresponding inbound. However, this effect may be delayed and is not guaranteed to be reflected
54+
* message-for-message.
55+
*
56+
* <p>What's actually guaranteed:
57+
*
58+
* <ol>
59+
* <li>If an application keeps writing messages to an outbound {@link StreamObserver#onNext} but
60+
* the peer stops consuming them (whether by not returning from the {@link
61+
* StreamObserver#onNext} callback or by not {@link #request}ing those callbacks), then
62+
* eventually the outbound's {@link #isReady} will become false and stay that way.
63+
* <li>Once in that state of backpressure, requesting and consuming enough messages from the
64+
* 'inbound' end will eventually cause the outbound to become ready again.
65+
* </ol>
66+
*
5067
* <p>DO NOT MOCK: The API is too complex to reliably mock. Use InProcessChannelBuilder to create
5168
* "real" RPCs suitable for testing.
5269
*
@@ -87,9 +104,10 @@ public abstract class CallStreamObserver<V> implements StreamObserver<V> {
87104
public abstract void setOnReadyHandler(Runnable onReadyHandler);
88105

89106
/**
90-
* Disables automatic flow control where a token is returned to the peer after a call
91-
* to the 'inbound' {@link io.grpc.stub.StreamObserver#onNext(Object)} has completed. If disabled
92-
* an application must make explicit calls to {@link #request} to receive messages.
107+
* Disables automatic flow control, a mode where another message is implicitly {@link #request}ed
108+
* after each call to the inbound's {@link StreamObserver#onNext(Object)} returns.
109+
*
110+
* <p>If disabled an application must make explicit calls to {@link #request} to receive messages.
93111
*
94112
* <p>On client-side this method may only be called during {@link
95113
* ClientResponseObserver#beforeStart}. On server-side it may only be called during the initial
@@ -116,8 +134,7 @@ public abstract class CallStreamObserver<V> implements StreamObserver<V> {
116134
public abstract void disableAutoInboundFlowControl();
117135

118136
/**
119-
* Requests the peer to produce {@code count} more messages to be delivered to the 'inbound'
120-
* {@link StreamObserver}.
137+
* Requests that {@code count} more messages be delivered to the 'inbound' {@link StreamObserver}.
121138
*
122139
* <p>This method is safe to call from multiple threads without external synchronization.
123140
*

stub/src/main/java/io/grpc/stub/ClientCallStreamObserver.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -92,8 +92,7 @@ public void disableAutoRequestWithInitial(int request) {
9292
public abstract void setOnReadyHandler(Runnable onReadyHandler);
9393

9494
/**
95-
* Requests the peer to produce {@code count} more messages to be delivered to the 'inbound'
96-
* {@link StreamObserver}.
95+
* Requests that {@code count} more messages be delivered to the 'inbound' {@link StreamObserver}.
9796
*
9897
* <p>This method is safe to call from multiple threads without external synchronization.
9998
*

stub/src/main/java/io/grpc/stub/ServerCallStreamObserver.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -147,8 +147,7 @@ public void disableAutoRequest() {
147147
public abstract void setOnReadyHandler(Runnable onReadyHandler);
148148

149149
/**
150-
* Requests the peer to produce {@code count} more messages to be delivered to the 'inbound'
151-
* {@link StreamObserver}.
150+
* Requests that {@code count} more messages be delivered to the 'inbound' {@link StreamObserver}.
152151
*
153152
* <p>This method is safe to call from multiple threads without external synchronization.
154153
*

0 commit comments

Comments
 (0)