#20418

Committed 24 Aug 2026 11:44PM UTC coverage: 89.185% (-0.02%) from 89.201%

Build # #20418

Build Type

push

github

Committed by

web-flow

Commit Message

stub: Clarify flow control javadoc (#12977)

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
buffer messages meaning the outbound stream can be isReady() even when
the receiver has no outstanding request()s for messages. Furthermore, those
buffers are sized in bytes and not all messages are 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.

Coverage Stats

38578 of 43256 relevant lines covered (89.19%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

20.0

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

/*
 * Copyright 2016 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.stub;

import static com.google.common.base.Preconditions.checkArgument;

import io.grpc.ExperimentalApi;

/**
 * A refinement of {@link CallStreamObserver} to allows for interaction with call
 * cancellation events on the server side. An instance of this class is obtained by casting the
 * {@code StreamObserver} passed as an argument to service implementations.
 *
 * <p>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.
 *
 * <p>DO NOT MOCK: The API is too complex to reliably mock. Use InProcessChannelBuilder to create
 * "real" RPCs suitable for testing and interact with the server using a normal client stub.
 */
public abstract class ServerCallStreamObserver<RespT> extends CallStreamObserver<RespT> {

  /**
   * Returns {@code true} when the call is cancelled and the server is encouraged to abort
   * processing to save resources, since the client will not be processing any further methods.
   * Cancellations can be caused by timeouts, explicit cancellation by client, network errors, and
   * similar.
   *
   * <p>This method may safely be called concurrently from multiple threads.
   */
  public abstract boolean isCancelled();

  /**
   * Sets a {@link Runnable} to be called if the call is cancelled and the server is encouraged to
   * abort processing to save resources, since the client will not process any further messages.
   * Cancellations can be caused by timeouts, explicit cancellation by the client, network errors,
   * etc.
   *
   * <p>It is guaranteed that execution of the {@link Runnable} is serialized with calls to the
   * 'inbound' {@link StreamObserver}. That also means that the callback will be delayed if other
   * callbacks are running; if one of those other callbacks runs for a significant amount of time
   * it can poll {@link #isCancelled()}, which is not delayed.
   *
   * <p>This method may only be called during the initial call to the application, before the
   * service returns its {@code StreamObserver}.
   *
   * <p>Setting the onCancelHandler will suppress the on-cancel exception thrown by
   * {@link #onNext}. If the caller is already handling cancellation via polling or cannot
   * substantially benefit from observing cancellation, using a no-op {@code onCancelHandler} is
   * useful just to suppress the {@code onNext()} exception.
   *
   * @param onCancelHandler to call when client has cancelled the call.
   */
  public abstract void setOnCancelHandler(Runnable onCancelHandler);


  /**
   * A hint to the call that specifies how many bytes must be queued before
   * {@link #isReady()} will return false. A call may ignore this property if
   * unsupported. This may only be set during stream initialization before
   * any messages are set.
   *
   * @param numBytes The number of bytes that must be queued. Must be a
   *                 positive integer.
   */
  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/11021")
  public void setOnReadyThreshold(int numBytes) {
    checkArgument(numBytes > 0, "numBytes must be positive: %s", numBytes);
  }

  /**
   * Sets the compression algorithm to use for the call. May only be called before sending any
   * messages. Default gRPC servers support the "gzip" compressor.
   *
   * <p>It is safe to call this even if the client does not support the compression format chosen.
   * The implementation will handle negotiation with the client and may fall back to no compression.
   *
   * @param compression the compression algorithm to use.
   * @throws IllegalArgumentException if the compressor name can not be found.
   */
  public abstract void setCompression(String compression);

  /**
   * Swaps to manual flow control where no message will be delivered to {@link
   * StreamObserver#onNext(Object)} unless it is {@link #request request()}ed.
   *
   * <p>It may only be called during the initial call to the application, before the service returns
   * its {@code StreamObserver}.
   *
   * <p>Note that for cases where the message is received before the service handler is invoked,
   * this method will have no effect. This is true for:
   *
   * <ul>
   *   <li>{@link io.grpc.MethodDescriptor.MethodType#UNARY} operations.</li>
   *   <li>{@link io.grpc.MethodDescriptor.MethodType#SERVER_STREAMING} operations.</li>
   * </ul>
   * </p>
   */
  public void disableAutoRequest() {
    throw new UnsupportedOperationException();
  }


  /**
   * If {@code true}, indicates that the observer is capable of sending additional messages
   * without requiring excessive buffering internally. This value is just a suggestion and the
   * application is free to ignore it, however doing so may result in excessive buffering within the
   * observer.
   *
   * <p>If {@code false}, the runnable passed to {@link #setOnReadyHandler} will be called after
   * {@code isReady()} transitions to {@code true}.
   */
  @Override
  public abstract boolean isReady();

  /**
   * Set a {@link Runnable} that will be executed every time the stream {@link #isReady()} state
   * changes from {@code false} to {@code true}.  While it is not guaranteed that the same
   * thread will always be used to execute the {@link Runnable}, it is guaranteed that executions
   * are serialized with calls to the 'inbound' {@link StreamObserver}.
   *
   * <p>May only be called during the initial call to the application, before the service returns
   * its {@code StreamObserver}.
   *
   * <p>Because there is a processing delay to deliver this notification, it is possible for
   * concurrent writes to cause {@code isReady() == false} within this callback. Handle "spurious"
   * notifications by checking {@code isReady()}'s current value instead of assuming it is now
   * {@code true}. If {@code isReady() == false} the normal expectations apply, so there would be
   * <em>another</em> {@code onReadyHandler} callback.
   *
   * @param onReadyHandler to call when peer is ready to receive more messages.
   */
  @Override
  public abstract void setOnReadyHandler(Runnable onReadyHandler);

  /**
   * Requests that {@code count} more messages be delivered to the 'inbound' {@link StreamObserver}.
   *
   * <p>This method is safe to call from multiple threads without external synchronization.
   *
   * @param count more messages
   */
  @Override
  public abstract void request(int count);

  /**
   * Sets message compression for subsequent calls to {@link #onNext}.
   *
   * @param enable whether to enable compression.
   */
  @Override
  public abstract void setMessageCompression(boolean enable);

  /**
   * Sets a {@link Runnable} to be executed when the call is closed cleanly from the server's
   * point of view: either {@link #onCompleted()} or {@link #onError(Throwable)} has been called,
   * all the messages and trailing metadata have been sent and the stream has been closed. Note
   * however that the client still may have not received all the messages due to network delay,
   * client crashes, and cancellation races.
   *
   * <p>Exactly one of {@code onCloseHandler} and {@code onCancelHandler} is guaranteed to be called
   * when the RPC terminates.</p>
   *
   * <p>It is guaranteed that execution of {@code onCloseHandler} is serialized with calls to
   * the 'inbound' {@link StreamObserver}. That also means that the callback will be delayed if
   * other callbacks are running.</p>
   *
   * <p>This method may only be called during the initial call to the application, before the
   * service returns its {@link StreamObserver request observer}.</p>
   *
   * @param onCloseHandler to execute when the call has been closed cleanly.
   */
  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/8467")
  public void setOnCloseHandler(Runnable onCloseHandler) {
    throw new UnsupportedOperationException();
  }
}

1	/*
2	* Copyright 2016 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.stub;
18
19	import static com.google.common.base.Preconditions.checkArgument;
20
21	import io.grpc.ExperimentalApi;
22
23	/**
24	* A refinement of {@link CallStreamObserver} to allows for interaction with call
25	* cancellation events on the server side. An instance of this class is obtained by casting the
26	* {@code StreamObserver} passed as an argument to service implementations.
27	*
28	* <p>Like {@code StreamObserver}, implementations are not required to be thread-safe; if multiple
29	* threads will be writing to an instance concurrently, the application must synchronize its calls.
30	*
31	* <p>DO NOT MOCK: The API is too complex to reliably mock. Use InProcessChannelBuilder to create
32	* "real" RPCs suitable for testing and interact with the server using a normal client stub.
33	*/
34	public abstract class ServerCallStreamObserver<RespT> extends CallStreamObserver<RespT> {	1 ✔
35
36	/**
37	* Returns {@code true} when the call is cancelled and the server is encouraged to abort
38	* processing to save resources, since the client will not be processing any further methods.
39	* Cancellations can be caused by timeouts, explicit cancellation by client, network errors, and
40	* similar.
41	*
42	* <p>This method may safely be called concurrently from multiple threads.
43	*/
44	public abstract boolean isCancelled();
45
46	/**
47	* Sets a {@link Runnable} to be called if the call is cancelled and the server is encouraged to
48	* abort processing to save resources, since the client will not process any further messages.
49	* Cancellations can be caused by timeouts, explicit cancellation by the client, network errors,
50	* etc.
51	*
52	* <p>It is guaranteed that execution of the {@link Runnable} is serialized with calls to the
53	* 'inbound' {@link StreamObserver}. That also means that the callback will be delayed if other
54	* callbacks are running; if one of those other callbacks runs for a significant amount of time
55	* it can poll {@link #isCancelled()}, which is not delayed.
56	*
57	* <p>This method may only be called during the initial call to the application, before the
58	* service returns its {@code StreamObserver}.
59	*
60	* <p>Setting the onCancelHandler will suppress the on-cancel exception thrown by
61	* {@link #onNext}. If the caller is already handling cancellation via polling or cannot
62	* substantially benefit from observing cancellation, using a no-op {@code onCancelHandler} is
63	* useful just to suppress the {@code onNext()} exception.
64	*
65	* @param onCancelHandler to call when client has cancelled the call.
66	*/
67	public abstract void setOnCancelHandler(Runnable onCancelHandler);
68
69
70	/**
71	* A hint to the call that specifies how many bytes must be queued before
72	* {@link #isReady()} will return false. A call may ignore this property if
73	* unsupported. This may only be set during stream initialization before
74	* any messages are set.
75	*
76	* @param numBytes The number of bytes that must be queued. Must be a
77	* positive integer.
78	*/
79	@ExperimentalApi("https://github.com/grpc/grpc-java/issues/11021")
80	public void setOnReadyThreshold(int numBytes) {
81	checkArgument(numBytes > 0, "numBytes must be positive: %s", numBytes);	×
82	}	×
83
84	/**
85	* Sets the compression algorithm to use for the call. May only be called before sending any
86	* messages. Default gRPC servers support the "gzip" compressor.
87	*
88	* <p>It is safe to call this even if the client does not support the compression format chosen.
89	* The implementation will handle negotiation with the client and may fall back to no compression.
90	*
91	* @param compression the compression algorithm to use.
92	* @throws IllegalArgumentException if the compressor name can not be found.
93	*/
94	public abstract void setCompression(String compression);
95
96	/**
97	* Swaps to manual flow control where no message will be delivered to {@link
98	* StreamObserver#onNext(Object)} unless it is {@link #request request()}ed.
99	*
100	* <p>It may only be called during the initial call to the application, before the service returns
101	* its {@code StreamObserver}.
102	*
103	* <p>Note that for cases where the message is received before the service handler is invoked,
104	* this method will have no effect. This is true for:
105	*
106	* <ul>
107	* <li>{@link io.grpc.MethodDescriptor.MethodType#UNARY} operations.</li>
108	* <li>{@link io.grpc.MethodDescriptor.MethodType#SERVER_STREAMING} operations.</li>
109	* </ul>
110	* </p>
111	*/
112	public void disableAutoRequest() {
113	throw new UnsupportedOperationException();	×
114	}
115
116
117	/**
118	* If {@code true}, indicates that the observer is capable of sending additional messages
119	* without requiring excessive buffering internally. This value is just a suggestion and the
120	* application is free to ignore it, however doing so may result in excessive buffering within the
121	* observer.
122	*
123	* <p>If {@code false}, the runnable passed to {@link #setOnReadyHandler} will be called after
124	* {@code isReady()} transitions to {@code true}.
125	*/
126	@Override
127	public abstract boolean isReady();
128
129	/**
130	* Set a {@link Runnable} that will be executed every time the stream {@link #isReady()} state
131	* changes from {@code false} to {@code true}. While it is not guaranteed that the same
132	* thread will always be used to execute the {@link Runnable}, it is guaranteed that executions
133	* are serialized with calls to the 'inbound' {@link StreamObserver}.
134	*
135	* <p>May only be called during the initial call to the application, before the service returns
136	* its {@code StreamObserver}.
137	*
138	* <p>Because there is a processing delay to deliver this notification, it is possible for
139	* concurrent writes to cause {@code isReady() == false} within this callback. Handle "spurious"
140	* notifications by checking {@code isReady()}'s current value instead of assuming it is now
141	* {@code true}. If {@code isReady() == false} the normal expectations apply, so there would be
142	* <em>another</em> {@code onReadyHandler} callback.
143	*
144	* @param onReadyHandler to call when peer is ready to receive more messages.
145	*/
146	@Override
147	public abstract void setOnReadyHandler(Runnable onReadyHandler);
148
149	/**
150	* Requests that {@code count} more messages be delivered to the 'inbound' {@link StreamObserver}.
151	*
152	* <p>This method is safe to call from multiple threads without external synchronization.
153	*
154	* @param count more messages
155	*/
156	@Override
157	public abstract void request(int count);
158
159	/**
160	* Sets message compression for subsequent calls to {@link #onNext}.
161	*
162	* @param enable whether to enable compression.
163	*/
164	@Override
165	public abstract void setMessageCompression(boolean enable);
166
167	/**
168	* Sets a {@link Runnable} to be executed when the call is closed cleanly from the server's
169	* point of view: either {@link #onCompleted()} or {@link #onError(Throwable)} has been called,
170	* all the messages and trailing metadata have been sent and the stream has been closed. Note
171	* however that the client still may have not received all the messages due to network delay,
172	* client crashes, and cancellation races.
173	*
174	* <p>Exactly one of {@code onCloseHandler} and {@code onCancelHandler} is guaranteed to be called
175	* when the RPC terminates.</p>
176	*
177	* <p>It is guaranteed that execution of {@code onCloseHandler} is serialized with calls to
178	* the 'inbound' {@link StreamObserver}. That also means that the callback will be delayed if
179	* other callbacks are running.</p>
180	*
181	* <p>This method may only be called during the initial call to the application, before the
182	* service returns its {@link StreamObserver request observer}.</p>
183	*
184	* @param onCloseHandler to execute when the call has been closed cleanly.
185	*/
186	@ExperimentalApi("https://github.com/grpc/grpc-java/issues/8467")
187	public void setOnCloseHandler(Runnable onCloseHandler) {
188	throw new UnsupportedOperationException();	×
189	}
190	}

grpc / grpc-java / #20418

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous