#20543

Committed 26 Oct 2026 12:31PM UTC coverage: 89.435% (-0.02%) from 89.451%

Build # #20543

Build Type

push

github

Committed by

web-flow

Commit Message

core: Reject retryThrottling without maxTokens or tokenRatio clearly (#13131)

## Problem

`ServiceConfigUtil.getThrottlePolicy()` reads the two required fields of
`retryThrottling` with `JsonUtil.getNumberAsDouble(...).floatValue()`.
When
either field is missing, `getNumberAsDouble` returns null and the call
fails
with an NPE:

Cannot invoke "java.lang.Double.floatValue()" because the return value
of
"io.grpc.internal.JsonUtil.getNumberAsDouble(java.util.Map, String)" is
null

`ScParser` catches it and the channel reports `UNKNOWN: failed to parse
service config`, but the cause doesn't say which field is missing. This
was
noted in the code as `TODO(dapengzhang0): check if this is null.`

## Fix

Check both fields with `checkNotNull(..., "<field> cannot be empty")`,
the
same message used for the required `retryPolicy` and `hedgingPolicy`
fields
in `ManagedChannelServiceConfig`. The exception type is still the NPE
that
`ScParser` already handles, so the only change is a clearer message.

## Testing

Added `getThrottlePolicy_failOnMissingMaxTokens` and
`getThrottlePolicy_failOnMissingTokenRatio` to `ServiceConfigUtilTest`.
Both
failed before the fix with the NPE above.

Coverage Stats

39396 of 44050 relevant lines covered (89.43%)

0.89 hits per line

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

96.0

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

/*
 * Copyright 2015 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 static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static io.grpc.internal.GrpcAttributes.ATTR_SECURITY_LEVEL;
import static io.grpc.internal.GrpcUtil.ACCEPT_ENCODING_SPLITTER;
import static io.grpc.internal.GrpcUtil.CONTENT_LENGTH_KEY;
import static io.grpc.internal.GrpcUtil.MESSAGE_ACCEPT_ENCODING_KEY;
import static io.grpc.internal.GrpcUtil.MESSAGE_ENCODING_KEY;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Throwables;
import com.google.common.util.concurrent.MoreExecutors;
import io.grpc.Attributes;
import io.grpc.Codec;
import io.grpc.Compressor;
import io.grpc.CompressorRegistry;
import io.grpc.Context;
import io.grpc.DecompressorRegistry;
import io.grpc.InternalDecompressorRegistry;
import io.grpc.InternalStatus;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.SecurityLevel;
import io.grpc.ServerCall;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.perfmark.PerfMark;
import io.perfmark.Tag;
import io.perfmark.TaskCloseable;
import java.io.InputStream;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;

final class ServerCallImpl<ReqT, RespT> extends ServerCall<ReqT, RespT> {

  private static final Logger log = Logger.getLogger(ServerCallImpl.class.getName());

  @VisibleForTesting
  static final String TOO_MANY_RESPONSES = "Too many responses";
  @VisibleForTesting
  static final String MISSING_RESPONSE = "Completed without a response";

  private final ServerStream stream;
  private final MethodDescriptor<ReqT, RespT> method;
  private final Tag tag;
  private final Context.CancellableContext context;
  private final byte[] messageAcceptEncoding;
  private final DecompressorRegistry decompressorRegistry;
  private final CompressorRegistry compressorRegistry;
  private CallTracer serverCallTracer;
  // The application's close() and the transport's closed() may race; only the first is counted.
  private final AtomicBoolean callEndReported = new AtomicBoolean();

  // state
  private volatile boolean cancelled;
  private boolean sendHeadersCalled;
  private boolean closeCalled;
  private Compressor compressor;
  private boolean messageSent;

  ServerCallImpl(ServerStream stream, MethodDescriptor<ReqT, RespT> method,
      Metadata inboundHeaders, Context.CancellableContext context,
      DecompressorRegistry decompressorRegistry, CompressorRegistry compressorRegistry,
      CallTracer serverCallTracer, Tag tag) {
    this.stream = stream;
    this.method = method;
    this.context = context;
    this.messageAcceptEncoding = inboundHeaders.get(MESSAGE_ACCEPT_ENCODING_KEY);
    this.decompressorRegistry = decompressorRegistry;
    this.compressorRegistry = compressorRegistry;
    this.serverCallTracer = serverCallTracer;
    this.serverCallTracer.reportCallStarted();
    this.tag = tag;
  }

  @Override
  public void request(int numMessages) {
    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.request")) {
      PerfMark.attachTag(tag);
      if (method.getType().clientSendsOneMessage() && numMessages == 1) {
        numMessages = 2;
      }
      stream.request(numMessages);
    }
  }

  @Override
  public void sendHeaders(Metadata headers) {
    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.sendHeaders")) {
      PerfMark.attachTag(tag);
      sendHeadersInternal(headers);
    }
  }

  private void sendHeadersInternal(Metadata headers) {
    checkState(!sendHeadersCalled, "sendHeaders has already been called");
    checkState(!closeCalled, "call is closed");

    headers.discardAll(CONTENT_LENGTH_KEY);
    headers.discardAll(MESSAGE_ENCODING_KEY);
    if (compressor == null) {
      compressor = Codec.Identity.NONE;
    } else {
      if (messageAcceptEncoding != null) {
        // TODO(carl-mastrangelo): remove the string allocation.
        if (!GrpcUtil.iterableContains(
            ACCEPT_ENCODING_SPLITTER.split(new String(messageAcceptEncoding, GrpcUtil.US_ASCII)),
            compressor.getMessageEncoding())) {
          // resort to using no compression.
          compressor = Codec.Identity.NONE;
        }
      } else {
        compressor = Codec.Identity.NONE;
      }
    }

    // Always put compressor, even if it's identity.
    headers.put(MESSAGE_ENCODING_KEY, compressor.getMessageEncoding());

    stream.setCompressor(compressor);

    headers.discardAll(MESSAGE_ACCEPT_ENCODING_KEY);
    byte[] advertisedEncodings =
        InternalDecompressorRegistry.getRawAdvertisedMessageEncodings(decompressorRegistry);
    if (advertisedEncodings.length != 0) {
      headers.put(MESSAGE_ACCEPT_ENCODING_KEY, advertisedEncodings);
    }

    // Don't check if sendMessage has been called, since it requires that sendHeaders was already
    // called.
    sendHeadersCalled = true;
    stream.writeHeaders(headers, !getMethodDescriptor().getType().serverSendsOneMessage());
  }

  @Override
  public void sendMessage(RespT message) {
    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.sendMessage")) {
      PerfMark.attachTag(tag);
      sendMessageInternal(message);
    }
  }

  private void sendMessageInternal(RespT message) {
    checkState(sendHeadersCalled, "sendHeaders has not been called");
    checkState(!closeCalled, "call is closed");

    if (method.getType().serverSendsOneMessage() && messageSent) {
      handleInternalError(Status.INTERNAL.withDescription(TOO_MANY_RESPONSES).asRuntimeException());
      return;
    }

    messageSent = true;
    try {
      InputStream resp = method.streamResponse(message);
      stream.writeMessage(resp);
      if (!getMethodDescriptor().getType().serverSendsOneMessage()) {
        stream.flush();
      }
    } catch (RuntimeException e) {
      handleInternalError(e);
    } catch (Error e) {
      close(
          Status.CANCELLED.withDescription("Server sendMessage() failed with Error"),
          new Metadata());
      throw e;
    }
  }

  @Override
  public void setMessageCompression(boolean enable) {
    stream.setMessageCompression(enable);
  }

  @Override
  public void setOnReadyThreshold(int numBytes) {
    stream.setOnReadyThreshold(numBytes);
  }

  @Override
  public void setCompression(String compressorName) {
    // Added here to give a better error message.
    checkState(!sendHeadersCalled, "sendHeaders has been called");

    compressor = compressorRegistry.lookupCompressor(compressorName);
    checkArgument(compressor != null, "Unable to find compressor by name %s", compressorName);
  }

  @Override
  public boolean isReady() {
    if (closeCalled) {
      return false;
    }
    return stream.isReady();
  }

  @Override
  public void close(Status status, Metadata trailers) {
    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.close")) {
      PerfMark.attachTag(tag);
      closeInternal(status, trailers);
    }
  }

  private void closeInternal(Status status, Metadata trailers) {
    checkState(!closeCalled, "call already closed");
    try {
      closeCalled = true;

      if (status.isOk() && method.getType().serverSendsOneMessage() && !messageSent) {
        handleInternalError(Status.INTERNAL.withDescription(MISSING_RESPONSE).asRuntimeException());
        return;
      }

      stream.close(status, trailers);
    } finally {
      reportCallEnded(status.isOk());
    }
  }

  private void reportCallEnded(boolean success) {
    if (callEndReported.compareAndSet(false, true)) {
      serverCallTracer.reportCallEnded(success);
    }
  }

  @Override
  public boolean isCancelled() {
    return cancelled;
  }

  ServerStreamListener newServerStreamListener(ServerCall.Listener<ReqT> listener) {
    return new ServerStreamListenerImpl<>(this, listener, context);
  }

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

  @Override
  public String getAuthority() {
    return stream.getAuthority();
  }

  @Override
  public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
    return method;
  }

  @Override
  public void triggerEvent(Object event) {
    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.triggerEvent")) {
      PerfMark.attachTag(tag);
      stream.triggerEvent(event);
    }
  }

  @Override
  public SecurityLevel getSecurityLevel() {
    final Attributes attributes = getAttributes();
    if (attributes == null) {
      return super.getSecurityLevel();
    }
    final SecurityLevel securityLevel = attributes.get(ATTR_SECURITY_LEVEL);
    return securityLevel == null ? super.getSecurityLevel() : securityLevel;
  }

  /**
   * Close the {@link ServerStream} because an internal error occurred. Allow the application to
   * run until completion, but silently ignore interactions with the {@link ServerStream} from now
   * on.
   */
  private void handleInternalError(Throwable internalError) {
    log.log(Level.WARNING, "Cancelling the stream because of internal error", internalError);
    Status status = (internalError instanceof StatusRuntimeException)
        ? ((StatusRuntimeException) internalError).getStatus()
        : Status.INTERNAL.withCause(internalError)
            .withDescription("Internal error so cancelling stream.");
    stream.cancel(status);
    reportCallEnded(false); // error so always false
  }

  /**
   * All of these callbacks are assumed to called on an application thread, and the caller is
   * responsible for handling thrown exceptions.
   */
  @VisibleForTesting
  static final class ServerStreamListenerImpl<ReqT> implements ServerStreamListener {
    private final ServerCallImpl<ReqT, ?> call;
    private final ServerCall.Listener<ReqT> listener;
    private final Context.CancellableContext context;
    private InputStream delayedMessage;

    public ServerStreamListenerImpl(
        ServerCallImpl<ReqT, ?> call, ServerCall.Listener<ReqT> listener,
        Context.CancellableContext context) {
      this.call = checkNotNull(call, "call");
      this.listener = checkNotNull(listener, "listener must not be null");
      this.context = checkNotNull(context, "context");
      // Wire ourselves up so that if the context is cancelled, our flag call.cancelled also
      // reflects the new state. Use a DirectExecutor so that it happens in the same thread
      // as the caller of {@link Context#cancel}.
      this.context.addListener(
          new Context.CancellationListener() {
            @Override
            public void cancelled(Context context) {
              // If the context has a cancellation cause then something exceptional happened
              // and we should also mark the call as cancelled.
              if (context.cancellationCause() != null) {
                ServerStreamListenerImpl.this.call.cancelled = true;
              }
            }
          },
          MoreExecutors.directExecutor());
    }

    @Override
    public void messagesAvailable(MessageProducer producer) {
      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.messagesAvailable")) {
        PerfMark.attachTag(call.tag);
        messagesAvailableInternal(producer);
      }
    }

    @SuppressWarnings("Finally") // The code avoids suppressing the exception thrown from try
    private void messagesAvailableInternal(final MessageProducer producer) {
      if (call.cancelled) {
        GrpcUtil.closeQuietly(producer);
        return;
      }

      InputStream message;
      try {
        while ((message = producer.next()) != null) {
          // TODO: Consider forcing this check to be done in the transport (MessageDeframer)
          // https://github.com/grpc/grpc-java/pull/13004/changes#r3939373996
          if (call.method.getType().clientSendsOneMessage()) {
            if (delayedMessage != null) {
              GrpcUtil.closeQuietly(message);
              call.stream.cancel(Status.INTERNAL.withDescription("Too many requests"));
              GrpcUtil.closeQuietly(delayedMessage);
              delayedMessage = null;
              call.cancelled = true;
              return;
            }
            delayedMessage = message;
          } else {
            try {
              listener.onMessage(call.method.parseRequest(message));
            } catch (Throwable t) {
              GrpcUtil.closeQuietly(message);
              throw t;
            }
            message.close();
          }
        }
      } catch (Throwable t) {
        GrpcUtil.closeQuietly(producer);
        Throwables.throwIfUnchecked(t);
        throw new RuntimeException(t);
      }
    }

    @Override
    public void halfClosed() {
      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.halfClosed")) {
        PerfMark.attachTag(call.tag);
        if (call.cancelled) {
          return;
        }

        if (delayedMessage != null) {
          InputStream message = delayedMessage;
          delayedMessage = null;
          try {
            listener.onMessage(call.method.parseRequest(message));
          } catch (Throwable t) {
            GrpcUtil.closeQuietly(message);
            Throwables.throwIfUnchecked(t);
            throw new RuntimeException(t);
          }
          GrpcUtil.closeQuietly(message);
        }

        listener.onHalfClose();
      }
    }

    @Override
    public void closed(Status status) {
      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.closed")) {
        PerfMark.attachTag(call.tag);
        closedInternal(status);
      }
    }

    private void closedInternal(Status status) {
      if (delayedMessage != null) {
        GrpcUtil.closeQuietly(delayedMessage);
        delayedMessage = null;
      }
      Throwable cancelCause = null;
      try {
        if (status.isOk()) {
          listener.onComplete();
        } else {
          call.cancelled = true;
          call.reportCallEnded(false);
          listener.onCancel();
          // The status will not have a cause in all failure scenarios, but we want to make sure
          // we always cancel the context with one to keep the context cancelled state consistent.
          cancelCause = InternalStatus.asRuntimeExceptionWithoutStacktrace(
              Status.CANCELLED.withDescription("RPC cancelled"), null);
        }
      } finally {
        // Cancel context after delivering RPC closure notification to allow the application to
        // clean up and update any state based on whether onComplete or onCancel was called.
        context.cancel(cancelCause);
      }
    }

    @Override
    public void onReady() {
      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.onReady")) {
        PerfMark.attachTag(call.tag);
        if (call.cancelled) {
          return;
        }
        listener.onReady();
      }
    }

    @Override
    public void triggerEvent(Object event) {
      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.triggerEvent")) {
        PerfMark.attachTag(call.tag);
        if (call.cancelled) {
          return;
        }
        listener.onEvent(event);
      }
    }
  }
}

1	/*
2	* Copyright 2015 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 static com.google.common.base.Preconditions.checkArgument;
20	import static com.google.common.base.Preconditions.checkNotNull;
21	import static com.google.common.base.Preconditions.checkState;
22	import static io.grpc.internal.GrpcAttributes.ATTR_SECURITY_LEVEL;
23	import static io.grpc.internal.GrpcUtil.ACCEPT_ENCODING_SPLITTER;
24	import static io.grpc.internal.GrpcUtil.CONTENT_LENGTH_KEY;
25	import static io.grpc.internal.GrpcUtil.MESSAGE_ACCEPT_ENCODING_KEY;
26	import static io.grpc.internal.GrpcUtil.MESSAGE_ENCODING_KEY;
27
28	import com.google.common.annotations.VisibleForTesting;
29	import com.google.common.base.Throwables;
30	import com.google.common.util.concurrent.MoreExecutors;
31	import io.grpc.Attributes;
32	import io.grpc.Codec;
33	import io.grpc.Compressor;
34	import io.grpc.CompressorRegistry;
35	import io.grpc.Context;
36	import io.grpc.DecompressorRegistry;
37	import io.grpc.InternalDecompressorRegistry;
38	import io.grpc.InternalStatus;
39	import io.grpc.Metadata;
40	import io.grpc.MethodDescriptor;
41	import io.grpc.SecurityLevel;
42	import io.grpc.ServerCall;
43	import io.grpc.Status;
44	import io.grpc.StatusRuntimeException;
45	import io.perfmark.PerfMark;
46	import io.perfmark.Tag;
47	import io.perfmark.TaskCloseable;
48	import java.io.InputStream;
49	import java.util.concurrent.atomic.AtomicBoolean;
50	import java.util.logging.Level;
51	import java.util.logging.Logger;
52
53	final class ServerCallImpl<ReqT, RespT> extends ServerCall<ReqT, RespT> {
54
55	private static final Logger log = Logger.getLogger(ServerCallImpl.class.getName());	1 ✔
56
57	@VisibleForTesting
58	static final String TOO_MANY_RESPONSES = "Too many responses";
59	@VisibleForTesting
60	static final String MISSING_RESPONSE = "Completed without a response";
61
62	private final ServerStream stream;
63	private final MethodDescriptor<ReqT, RespT> method;
64	private final Tag tag;
65	private final Context.CancellableContext context;
66	private final byte[] messageAcceptEncoding;
67	private final DecompressorRegistry decompressorRegistry;
68	private final CompressorRegistry compressorRegistry;
69	private CallTracer serverCallTracer;
70	// The application's close() and the transport's closed() may race; only the first is counted.
71	private final AtomicBoolean callEndReported = new AtomicBoolean();	1 ✔
72
73	// state
74	private volatile boolean cancelled;
75	private boolean sendHeadersCalled;
76	private boolean closeCalled;
77	private Compressor compressor;
78	private boolean messageSent;
79
80	ServerCallImpl(ServerStream stream, MethodDescriptor<ReqT, RespT> method,
81	Metadata inboundHeaders, Context.CancellableContext context,
82	DecompressorRegistry decompressorRegistry, CompressorRegistry compressorRegistry,
83	CallTracer serverCallTracer, Tag tag) {	1 ✔
84	this.stream = stream;	1 ✔
85	this.method = method;	1 ✔
86	this.context = context;	1 ✔
87	this.messageAcceptEncoding = inboundHeaders.get(MESSAGE_ACCEPT_ENCODING_KEY);	1 ✔
88	this.decompressorRegistry = decompressorRegistry;	1 ✔
89	this.compressorRegistry = compressorRegistry;	1 ✔
90	this.serverCallTracer = serverCallTracer;	1 ✔
91	this.serverCallTracer.reportCallStarted();	1 ✔
92	this.tag = tag;	1 ✔
93	}	1 ✔
94
95	@Override
96	public void request(int numMessages) {
97	try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.request")) {	1 ✔
98	PerfMark.attachTag(tag);	1 ✔
99	if (method.getType().clientSendsOneMessage() && numMessages == 1) {	1 ✔
100	numMessages = 2;	1 ✔
101	}
102	stream.request(numMessages);	1 ✔
103	}
104	}	1 ✔
105
106	@Override
107	public void sendHeaders(Metadata headers) {
108	try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.sendHeaders")) {	1 ✔
109	PerfMark.attachTag(tag);	1 ✔
110	sendHeadersInternal(headers);	1 ✔
111	}
112	}	1 ✔
113
114	private void sendHeadersInternal(Metadata headers) {
115	checkState(!sendHeadersCalled, "sendHeaders has already been called");	1 ✔
116	checkState(!closeCalled, "call is closed");	1 ✔
117
118	headers.discardAll(CONTENT_LENGTH_KEY);	1 ✔
119	headers.discardAll(MESSAGE_ENCODING_KEY);	1 ✔
120	if (compressor == null) {	1 ✔
121	compressor = Codec.Identity.NONE;	1 ✔
122	} else {
123	if (messageAcceptEncoding != null) {	1 ✔
124	// TODO(carl-mastrangelo): remove the string allocation.
125	if (!GrpcUtil.iterableContains(	1 ✔
126	ACCEPT_ENCODING_SPLITTER.split(new String(messageAcceptEncoding, GrpcUtil.US_ASCII)),	1 ✔
127	compressor.getMessageEncoding())) {	1 ✔
128	// resort to using no compression.
129	compressor = Codec.Identity.NONE;	1 ✔
130	}
131	} else {
132	compressor = Codec.Identity.NONE;	1 ✔
133	}
134	}
135
136	// Always put compressor, even if it's identity.
137	headers.put(MESSAGE_ENCODING_KEY, compressor.getMessageEncoding());	1 ✔
138
139	stream.setCompressor(compressor);	1 ✔
140
141	headers.discardAll(MESSAGE_ACCEPT_ENCODING_KEY);	1 ✔
142	byte[] advertisedEncodings =	1 ✔
143	InternalDecompressorRegistry.getRawAdvertisedMessageEncodings(decompressorRegistry);	1 ✔
144	if (advertisedEncodings.length != 0) {	1 ✔
145	headers.put(MESSAGE_ACCEPT_ENCODING_KEY, advertisedEncodings);	1 ✔
146	}
147
148	// Don't check if sendMessage has been called, since it requires that sendHeaders was already
149	// called.
150	sendHeadersCalled = true;	1 ✔
151	stream.writeHeaders(headers, !getMethodDescriptor().getType().serverSendsOneMessage());	1 ✔
152	}	1 ✔
153
154	@Override
155	public void sendMessage(RespT message) {
156	try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.sendMessage")) {	1 ✔
157	PerfMark.attachTag(tag);	1 ✔
158	sendMessageInternal(message);	1 ✔
159	}
160	}	1 ✔
161
162	private void sendMessageInternal(RespT message) {
163	checkState(sendHeadersCalled, "sendHeaders has not been called");	1 ✔
164	checkState(!closeCalled, "call is closed");	1 ✔
165
166	if (method.getType().serverSendsOneMessage() && messageSent) {	1 ✔
167	handleInternalError(Status.INTERNAL.withDescription(TOO_MANY_RESPONSES).asRuntimeException());	1 ✔
168	return;	1 ✔
169	}
170
171	messageSent = true;	1 ✔
172	try {
173	InputStream resp = method.streamResponse(message);	1 ✔
174	stream.writeMessage(resp);	1 ✔
175	if (!getMethodDescriptor().getType().serverSendsOneMessage()) {	1 ✔
176	stream.flush();	1 ✔
177	}
178	} catch (RuntimeException e) {	1 ✔
179	handleInternalError(e);	1 ✔
180	} catch (Error e) {	×
181	close(	×
182	Status.CANCELLED.withDescription("Server sendMessage() failed with Error"),	×
183	new Metadata());
184	throw e;	×
185	}	1 ✔
186	}	1 ✔
187
188	@Override
189	public void setMessageCompression(boolean enable) {
190	stream.setMessageCompression(enable);	1 ✔
191	}	1 ✔
192
193	@Override
194	public void setOnReadyThreshold(int numBytes) {
195	stream.setOnReadyThreshold(numBytes);	1 ✔
196	}	1 ✔
197
198	@Override
199	public void setCompression(String compressorName) {
200	// Added here to give a better error message.
201	checkState(!sendHeadersCalled, "sendHeaders has been called");	1 ✔
202
203	compressor = compressorRegistry.lookupCompressor(compressorName);	1 ✔
204	checkArgument(compressor != null, "Unable to find compressor by name %s", compressorName);	1 ✔
205	}	1 ✔
206
207	@Override
208	public boolean isReady() {
209	if (closeCalled) {	1 ✔
210	return false;	1 ✔
211	}
212	return stream.isReady();	1 ✔
213	}
214
215	@Override
216	public void close(Status status, Metadata trailers) {
217	try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.close")) {	1 ✔
218	PerfMark.attachTag(tag);	1 ✔
219	closeInternal(status, trailers);	1 ✔
220	}
221	}	1 ✔
222
223	private void closeInternal(Status status, Metadata trailers) {
224	checkState(!closeCalled, "call already closed");	1 ✔
225	try {
226	closeCalled = true;	1 ✔
227
228	if (status.isOk() && method.getType().serverSendsOneMessage() && !messageSent) {	1 ✔
229	handleInternalError(Status.INTERNAL.withDescription(MISSING_RESPONSE).asRuntimeException());	1 ✔
230	return;	1 ✔
231	}
232
233	stream.close(status, trailers);	1 ✔
234	} finally {
235	reportCallEnded(status.isOk());	1 ✔
236	}
237	}	1 ✔
238
239	private void reportCallEnded(boolean success) {
240	if (callEndReported.compareAndSet(false, true)) {	1 ✔
241	serverCallTracer.reportCallEnded(success);	1 ✔
242	}
243	}	1 ✔
244
245	@Override
246	public boolean isCancelled() {
247	return cancelled;	1 ✔
248	}
249
250	ServerStreamListener newServerStreamListener(ServerCall.Listener<ReqT> listener) {
251	return new ServerStreamListenerImpl<>(this, listener, context);	1 ✔
252	}
253
254	@Override
255	public Attributes getAttributes() {
256	return stream.getAttributes();	1 ✔
257	}
258
259	@Override
260	public String getAuthority() {
261	return stream.getAuthority();	1 ✔
262	}
263
264	@Override
265	public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
266	return method;	1 ✔
267	}
268
269	@Override
270	public void triggerEvent(Object event) {
271	try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.triggerEvent")) {	1 ✔
272	PerfMark.attachTag(tag);	1 ✔
273	stream.triggerEvent(event);	1 ✔
274	}
275	}	1 ✔
276
277	@Override
278	public SecurityLevel getSecurityLevel() {
279	final Attributes attributes = getAttributes();	1 ✔
280	if (attributes == null) {	1 ✔
281	return super.getSecurityLevel();	1 ✔
282	}
283	final SecurityLevel securityLevel = attributes.get(ATTR_SECURITY_LEVEL);	1 ✔
284	return securityLevel == null ? super.getSecurityLevel() : securityLevel;	1 ✔
285	}
286
287	/**
288	* Close the {@link ServerStream} because an internal error occurred. Allow the application to
289	* run until completion, but silently ignore interactions with the {@link ServerStream} from now
290	* on.
291	*/
292	private void handleInternalError(Throwable internalError) {
293	log.log(Level.WARNING, "Cancelling the stream because of internal error", internalError);	1 ✔
294	Status status = (internalError instanceof StatusRuntimeException)	1 ✔
295	? ((StatusRuntimeException) internalError).getStatus()	1 ✔
296	: Status.INTERNAL.withCause(internalError)	1 ✔
297	.withDescription("Internal error so cancelling stream.");	1 ✔
298	stream.cancel(status);	1 ✔
299	reportCallEnded(false); // error so always false	1 ✔
300	}	1 ✔
301
302	/**
303	* All of these callbacks are assumed to called on an application thread, and the caller is
304	* responsible for handling thrown exceptions.
305	*/
306	@VisibleForTesting
307	static final class ServerStreamListenerImpl<ReqT> implements ServerStreamListener {
308	private final ServerCallImpl<ReqT, ?> call;
309	private final ServerCall.Listener<ReqT> listener;
310	private final Context.CancellableContext context;
311	private InputStream delayedMessage;
312
313	public ServerStreamListenerImpl(
314	ServerCallImpl<ReqT, ?> call, ServerCall.Listener<ReqT> listener,
315	Context.CancellableContext context) {	1 ✔
316	this.call = checkNotNull(call, "call");	1 ✔
317	this.listener = checkNotNull(listener, "listener must not be null");	1 ✔
318	this.context = checkNotNull(context, "context");	1 ✔
319	// Wire ourselves up so that if the context is cancelled, our flag call.cancelled also
320	// reflects the new state. Use a DirectExecutor so that it happens in the same thread
321	// as the caller of {@link Context#cancel}.
322	this.context.addListener(	1 ✔
323	new Context.CancellationListener() {	1 ✔
324	@Override
325	public void cancelled(Context context) {
326	// If the context has a cancellation cause then something exceptional happened
327	// and we should also mark the call as cancelled.
328	if (context.cancellationCause() != null) {	1 ✔
329	ServerStreamListenerImpl.this.call.cancelled = true;	1 ✔
330	}
331	}	1 ✔
332	},
333	MoreExecutors.directExecutor());	1 ✔
334	}	1 ✔
335
336	@Override
337	public void messagesAvailable(MessageProducer producer) {
338	try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.messagesAvailable")) {	1 ✔
339	PerfMark.attachTag(call.tag);	1 ✔
340	messagesAvailableInternal(producer);	1 ✔
341	}
342	}	1 ✔
343
344	@SuppressWarnings("Finally") // The code avoids suppressing the exception thrown from try
345	private void messagesAvailableInternal(final MessageProducer producer) {
346	if (call.cancelled) {	1 ✔
347	GrpcUtil.closeQuietly(producer);	1 ✔
348	return;	1 ✔
349	}
350
351	InputStream message;
352	try {
353	while ((message = producer.next()) != null) {	1 ✔
354	// TODO: Consider forcing this check to be done in the transport (MessageDeframer)
355	// https://github.com/grpc/grpc-java/pull/13004/changes#r3939373996
356	if (call.method.getType().clientSendsOneMessage()) {	1 ✔
357	if (delayedMessage != null) {	1 ✔
358	GrpcUtil.closeQuietly(message);	1 ✔
359	call.stream.cancel(Status.INTERNAL.withDescription("Too many requests"));	1 ✔
360	GrpcUtil.closeQuietly(delayedMessage);	1 ✔
361	delayedMessage = null;	1 ✔
362	call.cancelled = true;	1 ✔
363	return;	1 ✔
364	}
365	delayedMessage = message;	1 ✔
366	} else {
367	try {
368	listener.onMessage(call.method.parseRequest(message));	1 ✔
369	} catch (Throwable t) {	1 ✔
370	GrpcUtil.closeQuietly(message);	1 ✔
371	throw t;	1 ✔
372	}	1 ✔
373	message.close();	1 ✔
374	}
375	}
376	} catch (Throwable t) {	1 ✔
377	GrpcUtil.closeQuietly(producer);	1 ✔
378	Throwables.throwIfUnchecked(t);	×
379	throw new RuntimeException(t);	×
380	}	1 ✔
381	}	1 ✔
382
383	@Override
384	public void halfClosed() {
385	try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.halfClosed")) {	1 ✔
386	PerfMark.attachTag(call.tag);	1 ✔
387	if (call.cancelled) {	1 ✔
388	return;	1 ✔
389	}
390
391	if (delayedMessage != null) {	1 ✔
392	InputStream message = delayedMessage;	1 ✔
393	delayedMessage = null;	1 ✔
394	try {
395	listener.onMessage(call.method.parseRequest(message));	1 ✔
396	} catch (Throwable t) {	1 ✔
397	GrpcUtil.closeQuietly(message);	1 ✔
398	Throwables.throwIfUnchecked(t);	×
399	throw new RuntimeException(t);	×
400	}	1 ✔
401	GrpcUtil.closeQuietly(message);	1 ✔
402	}
403
404	listener.onHalfClose();	1 ✔
405	}
406	}	1 ✔
407
408	@Override
409	public void closed(Status status) {
410	try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.closed")) {	1 ✔
411	PerfMark.attachTag(call.tag);	1 ✔
412	closedInternal(status);	1 ✔
413	}
414	}	1 ✔
415
416	private void closedInternal(Status status) {
417	if (delayedMessage != null) {	1 ✔
418	GrpcUtil.closeQuietly(delayedMessage);	1 ✔
419	delayedMessage = null;	1 ✔
420	}
421	Throwable cancelCause = null;	1 ✔
422	try {
423	if (status.isOk()) {	1 ✔
424	listener.onComplete();	1 ✔
425	} else {
426	call.cancelled = true;	1 ✔
427	call.reportCallEnded(false);	1 ✔
428	listener.onCancel();	1 ✔
429	// The status will not have a cause in all failure scenarios, but we want to make sure
430	// we always cancel the context with one to keep the context cancelled state consistent.
431	cancelCause = InternalStatus.asRuntimeExceptionWithoutStacktrace(	1 ✔
432	Status.CANCELLED.withDescription("RPC cancelled"), null);	1 ✔
433	}
434	} finally {
435	// Cancel context after delivering RPC closure notification to allow the application to
436	// clean up and update any state based on whether onComplete or onCancel was called.
437	context.cancel(cancelCause);	1 ✔
438	}
439	}	1 ✔
440
441	@Override
442	public void onReady() {
443	try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.onReady")) {	1 ✔
444	PerfMark.attachTag(call.tag);	1 ✔
445	if (call.cancelled) {	1 ✔
446	return;	1 ✔
447	}
448	listener.onReady();	1 ✔
449	}
450	}	1 ✔
451
452	@Override
453	public void triggerEvent(Object event) {
454	try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.triggerEvent")) {	1 ✔
455	PerfMark.attachTag(call.tag);	1 ✔
456	if (call.cancelled) {	1 ✔
457	return;	1 ✔
458	}
459	listener.onEvent(event);	1 ✔
460	}
461	}	1 ✔
462	}
463	}

grpc / grpc-java / #20543

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