#19971

Committed 01 Sep 2025 07:40AM UTC coverage: 88.568% (+0.02%) from 88.547%

Build # #19971

Build Type

push

github

Committed by

web-flow

Commit Message

allow java21 in jre matrix (#12281)

Run Details

34692 of 39170 relevant lines covered (88.57%)

0.89 hits per line

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

98.23

/../core/src/main/java/io/grpc/internal/AbstractServerStream.java

/*
 * Copyright 2014 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.internal;

import com.google.common.base.Preconditions;
import io.grpc.Attributes;
import io.grpc.Decompressor;
import io.grpc.InternalStatus;
import io.grpc.Metadata;
import io.grpc.Status;
import javax.annotation.Nullable;

/**
 * Abstract base class for {@link ServerStream} implementations. Extending classes only need to
 * implement {@link #transportState()} and {@link #abstractServerStreamSink()}. Must only be called
 * from the sending application thread.
 */
public abstract class AbstractServerStream extends AbstractStream
    implements ServerStream, MessageFramer.Sink {
  /**
   * A sink for outbound operations, separated from the stream simply to avoid name
   * collisions/confusion. Only called from application thread.
   */
  protected interface Sink {
    /**
     * Sends response headers to the remote end point.
     *
     * @param headers the headers to be sent to client.
     */
    void writeHeaders(Metadata headers, boolean flush);

    /**
     * Sends an outbound frame to the remote end point.
     *
     * @param frame a buffer containing the chunk of data to be sent.
     * @param flush {@code true} if more data may not be arriving soon
     * @param numMessages the number of messages this frame represents
     */
    void writeFrame(WritableBuffer frame, boolean flush, int numMessages);

    /**
     * Sends trailers to the remote end point. This call implies end of stream.
     *
     * @param trailers metadata to be sent to the end point
     * @param headersSent {@code true} if response headers have already been sent.
     * @param status the status that the call ended with
     */
    void writeTrailers(Metadata trailers, boolean headersSent, Status status);

    /**
     * Tears down the stream, typically in the event of a timeout. This method may be called
     * multiple times and from any thread.
     *
     * <p>This is a clone of {@link ServerStream#cancel(Status)}.
     */
    void cancel(Status status);
  }

  private final MessageFramer framer;
  private final StatsTraceContext statsTraceCtx;
  private boolean outboundClosed;
  private boolean headersSent;

  @SuppressWarnings("this-escape")
  protected AbstractServerStream(
      WritableBufferAllocator bufferAllocator, StatsTraceContext statsTraceCtx) {
    this.statsTraceCtx = Preconditions.checkNotNull(statsTraceCtx, "statsTraceCtx");
    this.framer = new MessageFramer(this, bufferAllocator, statsTraceCtx);
  }

  @Override
  protected abstract TransportState transportState();

  /**
   * Sink for transport to be called to perform outbound operations. Each stream must have its own
   * unique sink.
   */
  protected abstract Sink abstractServerStreamSink();

  @Override
  protected final MessageFramer framer() {
    return framer;
  }

  @Override
  public final void writeHeaders(Metadata headers, boolean flush) {
    Preconditions.checkNotNull(headers, "headers");

    headersSent = true;
    abstractServerStreamSink().writeHeaders(headers, flush);
  }

  @Override
  public final void deliverFrame(
      WritableBuffer frame, boolean endOfStream, boolean flush, int numMessages) {
    // Since endOfStream is triggered by the sending of trailers, avoid flush here and just flush
    // after the trailers.
    if (frame == null) {
      assert endOfStream;
      return;
    }
    if (endOfStream) {
      flush = false;
    }
    abstractServerStreamSink().writeFrame(frame, flush, numMessages);
  }

  @Override
  public final void close(Status status, Metadata trailers) {
    Preconditions.checkNotNull(status, "status");
    Preconditions.checkNotNull(trailers, "trailers");
    if (!outboundClosed) {
      outboundClosed = true;
      endOfMessages();
      addStatusToTrailers(trailers, status);
      // Safe to set without synchronization because access is tightly controlled.
      // closedStatus is only set from here, and is read from a place that has happen-after
      // guarantees with respect to here.
      transportState().setClosedStatus(status);
      abstractServerStreamSink().writeTrailers(trailers, headersSent, status);
    }
  }

  private void addStatusToTrailers(Metadata trailers, Status status) {
    trailers.discardAll(InternalStatus.CODE_KEY);
    trailers.discardAll(InternalStatus.MESSAGE_KEY);
    trailers.put(InternalStatus.CODE_KEY, status);
    if (status.getDescription() != null) {
      trailers.put(InternalStatus.MESSAGE_KEY, status.getDescription());
    }
  }

  @Override
  public final void cancel(Status status) {
    abstractServerStreamSink().cancel(status);
  }

  @Override
  public final boolean isReady() {
    return super.isReady();
  }

  @Override
  public final void setDecompressor(Decompressor decompressor) {
    transportState().setDecompressor(Preconditions.checkNotNull(decompressor, "decompressor"));
  }

  @Override public Attributes getAttributes() {
    return Attributes.EMPTY;
  }

  @Override
  public String getAuthority() {
    return null;
  }

  @Override
  public final void setListener(ServerStreamListener serverStreamListener) {
    transportState().setListener(serverStreamListener);
  }

  @Override
  public StatsTraceContext statsTraceContext() {
    return statsTraceCtx;
  }

  /**
   * A hint to the stream that specifies how many bytes must be queued before
   * {@link #isReady()} will return false. A stream may ignore this property
   * if unsupported. This may only be set before any messages are sent.
   *
   * @param numBytes The number of bytes that must be queued. Must be a
   *                 positive integer.
   */
  @Override
  public void setOnReadyThreshold(int numBytes) {
    super.setOnReadyThreshold(numBytes);
  }

  /**
   * This should only be called from the transport thread (except for private interactions with
   * {@code AbstractServerStream}).
   */
  protected abstract static class TransportState extends AbstractStream.TransportState {
    /** Whether listener.closed() has been called. */
    private boolean listenerClosed;
    private ServerStreamListener listener;
    private final StatsTraceContext statsTraceCtx;

    private boolean endOfStream = false;
    private boolean deframerClosed = false;
    private boolean immediateCloseRequested = false;
    private Runnable deframerClosedTask;
    /** The status that the application used to close this stream. */
    @Nullable
    private Status closedStatus;

    protected TransportState(
        int maxMessageSize,
        StatsTraceContext statsTraceCtx,
        TransportTracer transportTracer) {
      super(
          maxMessageSize,
          statsTraceCtx,
          Preconditions.checkNotNull(transportTracer, "transportTracer"));
      this.statsTraceCtx = Preconditions.checkNotNull(statsTraceCtx, "statsTraceCtx");
    }

    /**
     * Sets the listener to receive notifications. Must be called in the context of the transport
     * thread.
     */
    public final void setListener(ServerStreamListener listener) {
      Preconditions.checkState(this.listener == null, "setListener should be called only once");
      this.listener = Preconditions.checkNotNull(listener, "listener");
    }

    @Override
    public final void onStreamAllocated() {
      super.onStreamAllocated();
      getTransportTracer().reportRemoteStreamStarted();
    }

    @Override
    public void deframerClosed(boolean hasPartialMessage) {
      deframerClosed = true;
      if (endOfStream && !immediateCloseRequested) {
        if (hasPartialMessage) {
          // We've received the entire stream and have data available but we don't have
          // enough to read the next frame ... this is bad.
          deframeFailed(
              Status.INTERNAL
                  .withDescription("Encountered end-of-stream mid-frame")
                  .asRuntimeException());
          deframerClosedTask = null;
          return;
        }
        listener.halfClosed();
      }
      if (deframerClosedTask != null) {
        deframerClosedTask.run();
        deframerClosedTask = null;
      }
    }



    @Override
    protected ServerStreamListener listener() {
      return listener;
    }

    /**
     * Called in the transport thread to process the content of an inbound DATA frame from the
     * client.
     *
     * @param frame the inbound HTTP/2 DATA frame. If this buffer is not used immediately, it must
     *              be retained.
     * @param endOfStream {@code true} if no more data will be received on the stream.
     */
    public void inboundDataReceived(ReadableBuffer frame, boolean endOfStream) {
      Preconditions.checkState(!this.endOfStream, "Past end of stream");
      // Deframe the message. If a failure occurs, deframeFailed will be called.
      deframe(frame);
      if (endOfStream) {
        this.endOfStream = true;
        closeDeframer(false);
      }
    }

    /**
     * Notifies failure to the listener of the stream. The transport is responsible for notifying
     * the client of the failure independent of this method.
     *
     * <p>Unlike {@link #close(Status, Metadata)}, this method is only called from the
     * transport. The transport should use this method instead of {@code close(Status)} for internal
     * errors to prevent exposing unexpected states and exceptions to the application.
     *
     * @param status the error status. Must not be {@link Status#OK}.
     */
    public final void transportReportStatus(final Status status) {
      Preconditions.checkArgument(!status.isOk(), "status must not be OK");
      onStreamDeallocated();
      if (deframerClosed) {
        deframerClosedTask = null;
        closeListener(status);
      } else {
        deframerClosedTask =
            new Runnable() {
              @Override
              public void run() {
                closeListener(status);
              }
            };
        immediateCloseRequested = true;
        closeDeframer(true);
      }
    }

    /**
     * Indicates the stream is considered completely closed and there is no further opportunity for
     * error. It calls the listener's {@code closed()} if it was not already done by {@link
     * #transportReportStatus}.
     */
    public void complete() {
      onStreamDeallocated();
      if (deframerClosed) {
        deframerClosedTask = null;
        closeListener(Status.OK);
      } else {
        deframerClosedTask =
            new Runnable() {
              @Override
              public void run() {
                closeListener(Status.OK);
              }
            };
        immediateCloseRequested = true;
        closeDeframer(true);
      }
    }

    /**
     * Closes the listener if not previously closed and frees resources. {@code newStatus} is a
     * status generated by gRPC. It is <b>not</b> the status the stream closed with.
     */
    private void closeListener(Status newStatus) {
      // If newStatus is OK, the application must have already called AbstractServerStream.close()
      // and the status passed in there was the actual status of the RPC.
      // If newStatus non-OK, then the RPC ended some other way and the server application did
      // not initiate the termination.
      Preconditions.checkState(!newStatus.isOk() || closedStatus != null);
      if (!listenerClosed) {
        if (!newStatus.isOk()) {
          statsTraceCtx.streamClosed(newStatus);
          getTransportTracer().reportStreamClosed(false);
        } else {
          statsTraceCtx.streamClosed(closedStatus);
          getTransportTracer().reportStreamClosed(closedStatus.isOk());
        }
        listenerClosed = true;
        listener().closed(newStatus);
      }
    }

    /**
     * Stores the {@code Status} that the application used to close this stream.
     */
    private void setClosedStatus(Status closeStatus) {
      Preconditions.checkState(closedStatus == null, "closedStatus can only be set once");
      closedStatus = closeStatus;
    }
  }
}

1	/*
2	* Copyright 2014 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.internal;
18
19	import com.google.common.base.Preconditions;
20	import io.grpc.Attributes;
21	import io.grpc.Decompressor;
22	import io.grpc.InternalStatus;
23	import io.grpc.Metadata;
24	import io.grpc.Status;
25	import javax.annotation.Nullable;
26
27	/**
28	* Abstract base class for {@link ServerStream} implementations. Extending classes only need to
29	* implement {@link #transportState()} and {@link #abstractServerStreamSink()}. Must only be called
30	* from the sending application thread.
31	*/
32	public abstract class AbstractServerStream extends AbstractStream	1✔
33	implements ServerStream, MessageFramer.Sink {
34	/**
35	* A sink for outbound operations, separated from the stream simply to avoid name
36	* collisions/confusion. Only called from application thread.
37	*/
38	protected interface Sink {
39	/**
40	* Sends response headers to the remote end point.
41	*
42	* @param headers the headers to be sent to client.
43	*/
44	void writeHeaders(Metadata headers, boolean flush);
45
46	/**
47	* Sends an outbound frame to the remote end point.
48	*
49	* @param frame a buffer containing the chunk of data to be sent.
50	* @param flush {@code true} if more data may not be arriving soon
51	* @param numMessages the number of messages this frame represents
52	*/
53	void writeFrame(WritableBuffer frame, boolean flush, int numMessages);
54
55	/**
56	* Sends trailers to the remote end point. This call implies end of stream.
57	*
58	* @param trailers metadata to be sent to the end point
59	* @param headersSent {@code true} if response headers have already been sent.
60	* @param status the status that the call ended with
61	*/
62	void writeTrailers(Metadata trailers, boolean headersSent, Status status);
63
64	/**
65	* Tears down the stream, typically in the event of a timeout. This method may be called
66	* multiple times and from any thread.
67	*
68	* <p>This is a clone of {@link ServerStream#cancel(Status)}.
69	*/
70	void cancel(Status status);
71	}
72
73	private final MessageFramer framer;
74	private final StatsTraceContext statsTraceCtx;
75	private boolean outboundClosed;
76	private boolean headersSent;
77
78	@SuppressWarnings("this-escape")
79	protected AbstractServerStream(
80	WritableBufferAllocator bufferAllocator, StatsTraceContext statsTraceCtx) {	1✔
81	this.statsTraceCtx = Preconditions.checkNotNull(statsTraceCtx, "statsTraceCtx");	1✔
82	this.framer = new MessageFramer(this, bufferAllocator, statsTraceCtx);	1✔
83	}	1✔
84
85	@Override
86	protected abstract TransportState transportState();
87
88	/**
89	* Sink for transport to be called to perform outbound operations. Each stream must have its own
90	* unique sink.
91	*/
92	protected abstract Sink abstractServerStreamSink();
93
94	@Override
95	protected final MessageFramer framer() {
96	return framer;	1✔
97	}
98
99	@Override
100	public final void writeHeaders(Metadata headers, boolean flush) {
101	Preconditions.checkNotNull(headers, "headers");	1✔
102
103	headersSent = true;	1✔
104	abstractServerStreamSink().writeHeaders(headers, flush);	1✔
105	}	1✔
106
107	@Override
108	public final void deliverFrame(
109	WritableBuffer frame, boolean endOfStream, boolean flush, int numMessages) {
110	// Since endOfStream is triggered by the sending of trailers, avoid flush here and just flush
111	// after the trailers.
112	if (frame == null) {	1✔
113	assert endOfStream;	1✔
114	return;	1✔
115	}
116	if (endOfStream) {	1✔
117	flush = false;	1✔
118	}
119	abstractServerStreamSink().writeFrame(frame, flush, numMessages);	1✔
120	}	1✔
121
122	@Override
123	public final void close(Status status, Metadata trailers) {
124	Preconditions.checkNotNull(status, "status");	1✔
125	Preconditions.checkNotNull(trailers, "trailers");	1✔
126	if (!outboundClosed) {	1✔
127	outboundClosed = true;	1✔
128	endOfMessages();	1✔
129	addStatusToTrailers(trailers, status);	1✔
130	// Safe to set without synchronization because access is tightly controlled.
131	// closedStatus is only set from here, and is read from a place that has happen-after
132	// guarantees with respect to here.
133	transportState().setClosedStatus(status);	1✔
134	abstractServerStreamSink().writeTrailers(trailers, headersSent, status);	1✔
135	}
136	}	1✔
137
138	private void addStatusToTrailers(Metadata trailers, Status status) {
139	trailers.discardAll(InternalStatus.CODE_KEY);	1✔
140	trailers.discardAll(InternalStatus.MESSAGE_KEY);	1✔
141	trailers.put(InternalStatus.CODE_KEY, status);	1✔
142	if (status.getDescription() != null) {	1✔
143	trailers.put(InternalStatus.MESSAGE_KEY, status.getDescription());	1✔
144	}
145	}	1✔
146
147	@Override
148	public final void cancel(Status status) {
149	abstractServerStreamSink().cancel(status);	1✔
150	}	1✔
151
152	@Override
153	public final boolean isReady() {
154	return super.isReady();	1✔
155	}
156
157	@Override
158	public final void setDecompressor(Decompressor decompressor) {
159	transportState().setDecompressor(Preconditions.checkNotNull(decompressor, "decompressor"));	1✔
160	}	1✔
161
162	@Override public Attributes getAttributes() {
163	return Attributes.EMPTY;	×
164	}
165
166	@Override
167	public String getAuthority() {
168	return null;	×
169	}
170
171	@Override
172	public final void setListener(ServerStreamListener serverStreamListener) {
173	transportState().setListener(serverStreamListener);	1✔
174	}	1✔
175
176	@Override
177	public StatsTraceContext statsTraceContext() {
178	return statsTraceCtx;	1✔
179	}
180
181	/**
182	* A hint to the stream that specifies how many bytes must be queued before
183	* {@link #isReady()} will return false. A stream may ignore this property
184	* if unsupported. This may only be set before any messages are sent.
185	*
186	* @param numBytes The number of bytes that must be queued. Must be a
187	* positive integer.
188	*/
189	@Override
190	public void setOnReadyThreshold(int numBytes) {
191	super.setOnReadyThreshold(numBytes);	1✔
192	}	1✔
193
194	/**
195	* This should only be called from the transport thread (except for private interactions with
196	* {@code AbstractServerStream}).
197	*/
198	protected abstract static class TransportState extends AbstractStream.TransportState {
199	/** Whether listener.closed() has been called. */
200	private boolean listenerClosed;
201	private ServerStreamListener listener;
202	private final StatsTraceContext statsTraceCtx;
203
204	private boolean endOfStream = false;	1✔
205	private boolean deframerClosed = false;	1✔
206	private boolean immediateCloseRequested = false;	1✔
207	private Runnable deframerClosedTask;
208	/** The status that the application used to close this stream. */
209	@Nullable
210	private Status closedStatus;
211
212	protected TransportState(
213	int maxMessageSize,
214	StatsTraceContext statsTraceCtx,
215	TransportTracer transportTracer) {
216	super(	1✔
217	maxMessageSize,
218	statsTraceCtx,
219	Preconditions.checkNotNull(transportTracer, "transportTracer"));	1✔
220	this.statsTraceCtx = Preconditions.checkNotNull(statsTraceCtx, "statsTraceCtx");	1✔
221	}	1✔
222
223	/**
224	* Sets the listener to receive notifications. Must be called in the context of the transport
225	* thread.
226	*/
227	public final void setListener(ServerStreamListener listener) {
228	Preconditions.checkState(this.listener == null, "setListener should be called only once");	1✔
229	this.listener = Preconditions.checkNotNull(listener, "listener");	1✔
230	}	1✔
231
232	@Override
233	public final void onStreamAllocated() {
234	super.onStreamAllocated();	1✔
235	getTransportTracer().reportRemoteStreamStarted();	1✔
236	}	1✔
237
238	@Override
239	public void deframerClosed(boolean hasPartialMessage) {
240	deframerClosed = true;	1✔
241	if (endOfStream && !immediateCloseRequested) {	1✔
242	if (hasPartialMessage) {	1✔
243	// We've received the entire stream and have data available but we don't have
244	// enough to read the next frame ... this is bad.
245	deframeFailed(	1✔
246	Status.INTERNAL
247	.withDescription("Encountered end-of-stream mid-frame")	1✔
248	.asRuntimeException());	1✔
249	deframerClosedTask = null;	1✔
250	return;	1✔
251	}
252	listener.halfClosed();	1✔
253	}
254	if (deframerClosedTask != null) {	1✔
255	deframerClosedTask.run();	1✔
256	deframerClosedTask = null;	1✔
257	}
258	}	1✔
259
260
261
262	@Override
263	protected ServerStreamListener listener() {
264	return listener;	1✔
265	}
266
267	/**
268	* Called in the transport thread to process the content of an inbound DATA frame from the
269	* client.
270	*
271	* @param frame the inbound HTTP/2 DATA frame. If this buffer is not used immediately, it must
272	* be retained.
273	* @param endOfStream {@code true} if no more data will be received on the stream.
274	*/
275	public void inboundDataReceived(ReadableBuffer frame, boolean endOfStream) {
276	Preconditions.checkState(!this.endOfStream, "Past end of stream");	1✔
277	// Deframe the message. If a failure occurs, deframeFailed will be called.
278	deframe(frame);	1✔
279	if (endOfStream) {	1✔
280	this.endOfStream = true;	1✔
281	closeDeframer(false);	1✔
282	}
283	}	1✔
284
285	/**
286	* Notifies failure to the listener of the stream. The transport is responsible for notifying
287	* the client of the failure independent of this method.
288	*
289	* <p>Unlike {@link #close(Status, Metadata)}, this method is only called from the
290	* transport. The transport should use this method instead of {@code close(Status)} for internal
291	* errors to prevent exposing unexpected states and exceptions to the application.
292	*
293	* @param status the error status. Must not be {@link Status#OK}.
294	*/
295	public final void transportReportStatus(final Status status) {
296	Preconditions.checkArgument(!status.isOk(), "status must not be OK");	1✔
297	onStreamDeallocated();	1✔
298	if (deframerClosed) {	1✔
299	deframerClosedTask = null;	1✔
300	closeListener(status);	1✔
301	} else {
302	deframerClosedTask =	1✔
303	new Runnable() {	1✔
304	@Override
305	public void run() {
306	closeListener(status);	1✔
307	}	1✔
308	};
309	immediateCloseRequested = true;	1✔
310	closeDeframer(true);	1✔
311	}
312	}	1✔
313
314	/**
315	* Indicates the stream is considered completely closed and there is no further opportunity for
316	* error. It calls the listener's {@code closed()} if it was not already done by {@link
317	* #transportReportStatus}.
318	*/
319	public void complete() {
320	onStreamDeallocated();	1✔
321	if (deframerClosed) {	1✔
322	deframerClosedTask = null;	1✔
323	closeListener(Status.OK);	1✔
324	} else {
325	deframerClosedTask =	1✔
326	new Runnable() {	1✔
327	@Override
328	public void run() {
329	closeListener(Status.OK);	1✔
330	}	1✔
331	};
332	immediateCloseRequested = true;	1✔
333	closeDeframer(true);	1✔
334	}
335	}	1✔
336
337	/**
338	* Closes the listener if not previously closed and frees resources. {@code newStatus} is a
339	* status generated by gRPC. It is <b>not</b> the status the stream closed with.
340	*/
341	private void closeListener(Status newStatus) {
342	// If newStatus is OK, the application must have already called AbstractServerStream.close()
343	// and the status passed in there was the actual status of the RPC.
344	// If newStatus non-OK, then the RPC ended some other way and the server application did
345	// not initiate the termination.
346	Preconditions.checkState(!newStatus.isOk() \|\| closedStatus != null);	1✔
347	if (!listenerClosed) {	1✔
348	if (!newStatus.isOk()) {	1✔
349	statsTraceCtx.streamClosed(newStatus);	1✔
350	getTransportTracer().reportStreamClosed(false);	1✔
351	} else {
352	statsTraceCtx.streamClosed(closedStatus);	1✔
353	getTransportTracer().reportStreamClosed(closedStatus.isOk());	1✔
354	}
355	listenerClosed = true;	1✔
356	listener().closed(newStatus);	1✔
357	}
358	}	1✔
359
360	/**
361	* Stores the {@code Status} that the application used to close this stream.
362	*/
363	private void setClosedStatus(Status closeStatus) {
364	Preconditions.checkState(closedStatus == null, "closedStatus can only be set once");	1✔
365	closedStatus = closeStatus;	1✔
366	}	1✔
367	}
368	}

grpc / grpc-java / #19971

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