From b0eb447c5806cba8dec35258adcbe813e2086b3b Mon Sep 17 00:00:00 2001 From: jdcormie Date: Fri, 7 Aug 2026 11:24:52 -0700 Subject: [PATCH 1/2] 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 a single request() may not cause a non-ready outbound to become ready again, because freeing a single variably-sized message from the network buffer does not guarantee the sender's window will drop below its ready threshold. Replace this with a short discussion of what is actually guaranteed by every transport. TAG=agy CONV=ea489767-38f7-42d6-9245-87a9e1e71f35 --- .../java/io/grpc/stub/CallStreamObserver.java | 20 +++++++++++++------ .../grpc/stub/ClientCallStreamObserver.java | 3 +-- .../grpc/stub/ServerCallStreamObserver.java | 3 +-- 3 files changed, 16 insertions(+), 10 deletions(-) diff --git a/stub/src/main/java/io/grpc/stub/CallStreamObserver.java b/stub/src/main/java/io/grpc/stub/CallStreamObserver.java index 52dd046831d..de10d321e4e 100644 --- a/stub/src/main/java/io/grpc/stub/CallStreamObserver.java +++ b/stub/src/main/java/io/grpc/stub/CallStreamObserver.java @@ -47,6 +47,14 @@ *

Like {@code StreamObserver}, implementations are not required to be thread-safe; if multiple * threads will be writing to an instance concurrently, the application must synchronize its calls. * + *

On flow control: {@link #isReady} returns whether an observer can accept more messages without + * excessive buffering. Readiness is affected by the volume of messages previously sent and how much + * of that data has been requested by the peer application. But the link between local readiness and + * peer message requests may not be immediate and may not be reflected message-for-message. All + * that's guaranteed is this: if an outbound is being fed messages but the peer is not requesting + * them, the outbound's {@link #isReady} will eventually return false. If, later, the peer does + * request enough of those messages, the outbound observer will eventually become ready again. + * *

DO NOT MOCK: The API is too complex to reliably mock. Use InProcessChannelBuilder to create * "real" RPCs suitable for testing. * @@ -87,9 +95,10 @@ public abstract class CallStreamObserver implements StreamObserver { public abstract void setOnReadyHandler(Runnable onReadyHandler); /** - * Disables automatic flow control where a token is returned to the peer after a call - * to the 'inbound' {@link io.grpc.stub.StreamObserver#onNext(Object)} has completed. If disabled - * an application must make explicit calls to {@link #request} to receive messages. + * Disables automatic flow control, a mode where another message is implicitly {@link #request}ed + * after each call to the inbound's {@link StreamObserver#onNext(Object)} returns. + * + *

If disabled an application must make explicit calls to {@link #request} to receive messages. * *

On client-side this method may only be called during {@link * ClientResponseObserver#beforeStart}. On server-side it may only be called during the initial @@ -108,7 +117,7 @@ public abstract class CallStreamObserver implements StreamObserver { * * *

- * + * *

This API is being replaced, but is not yet deprecated. On server-side it being replaced * with {@link ServerCallStreamObserver#disableAutoRequest}. On client-side {@link * ClientCallStreamObserver#disableAutoRequestWithInitial disableAutoRequestWithInitial(1)}. @@ -116,8 +125,7 @@ public abstract class CallStreamObserver implements StreamObserver { public abstract void disableAutoInboundFlowControl(); /** - * Requests the peer to produce {@code count} more messages to be delivered to the 'inbound' - * {@link StreamObserver}. + * Requests that {@code count} more messages be delivered to the 'inbound' {@link StreamObserver}. * *

This method is safe to call from multiple threads without external synchronization. * diff --git a/stub/src/main/java/io/grpc/stub/ClientCallStreamObserver.java b/stub/src/main/java/io/grpc/stub/ClientCallStreamObserver.java index 8f420fa77e4..4a173520359 100644 --- a/stub/src/main/java/io/grpc/stub/ClientCallStreamObserver.java +++ b/stub/src/main/java/io/grpc/stub/ClientCallStreamObserver.java @@ -92,8 +92,7 @@ public void disableAutoRequestWithInitial(int request) { public abstract void setOnReadyHandler(Runnable onReadyHandler); /** - * Requests the peer to produce {@code count} more messages to be delivered to the 'inbound' - * {@link StreamObserver}. + * Requests that {@code count} more messages be delivered to the 'inbound' {@link StreamObserver}. * *

This method is safe to call from multiple threads without external synchronization. * diff --git a/stub/src/main/java/io/grpc/stub/ServerCallStreamObserver.java b/stub/src/main/java/io/grpc/stub/ServerCallStreamObserver.java index 6ffea3500cc..cc5fbaf3fb0 100644 --- a/stub/src/main/java/io/grpc/stub/ServerCallStreamObserver.java +++ b/stub/src/main/java/io/grpc/stub/ServerCallStreamObserver.java @@ -147,8 +147,7 @@ public void disableAutoRequest() { public abstract void setOnReadyHandler(Runnable onReadyHandler); /** - * Requests the peer to produce {@code count} more messages to be delivered to the 'inbound' - * {@link StreamObserver}. + * Requests that {@code count} more messages be delivered to the 'inbound' {@link StreamObserver}. * *

This method is safe to call from multiple threads without external synchronization. * From 8edaf8aaa2a6f4bf44da3ca658957d3fbc9e3fe1 Mon Sep 17 00:00:00 2001 From: jdcormie Date: Mon, 24 Aug 2026 14:34:42 -0700 Subject: [PATCH 2/2] Address two in-person review comments. Avoid the word "link" because it means something else in networking. Avoid the word "requests" because other grpc language libraries don't have a request method or even the same concept. --- .../src/main/java/io/grpc/stub/CallStreamObserver.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/stub/src/main/java/io/grpc/stub/CallStreamObserver.java b/stub/src/main/java/io/grpc/stub/CallStreamObserver.java index de10d321e4e..f91d2d2450a 100644 --- a/stub/src/main/java/io/grpc/stub/CallStreamObserver.java +++ b/stub/src/main/java/io/grpc/stub/CallStreamObserver.java @@ -49,11 +49,11 @@ * *

On flow control: {@link #isReady} returns whether an observer can accept more messages without * excessive buffering. Readiness is affected by the volume of messages previously sent and how much - * of that data has been requested by the peer application. But the link between local readiness and - * peer message requests may not be immediate and may not be reflected message-for-message. All - * that's guaranteed is this: if an outbound is being fed messages but the peer is not requesting - * them, the outbound's {@link #isReady} will eventually return false. If, later, the peer does - * request enough of those messages, the outbound observer will eventually become ready again. + * of that data has been requested by the peer application. But this effect may not be immediate and + * may not be message-for-message. All that's guaranteed is this: if an outbound is being fed + * messages but the peer is not requesting them, the outbound's {@link #isReady} will eventually + * return false. If, later, the peer does ask for enough of those messages, the outbound observer + * will eventually become ready again. * *

DO NOT MOCK: The API is too complex to reliably mock. Use InProcessChannelBuilder to create * "real" RPCs suitable for testing.