import io.grpc.Status.Code; //导入依赖的package包/类
@Override
public boolean tableExists(String tableId) throws IOException {
  try (BigtableSession session = new BigtableSession(options)) {
    GetTableRequest getTable =
        GetTableRequest.newBuilder()
            .setName(options.getInstanceName().toTableNameStr(tableId))
            .build();
    session.getTableAdminClient().getTable(getTable);
    return true;
  } catch (StatusRuntimeException e) {
    if (e.getStatus().getCode() == Code.NOT_FOUND) {
      return false;
    }
    String message =
        String.format(
            "Error checking whether table %s (BigtableOptions %s) exists", tableId, options);
    LOG.error(message, e);
    throw new IOException(message, e);
  }
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void grpcServerGetsStopped() {
    final DropwizardTestSupport<TestConfiguration> testSupport =
            new DropwizardTestSupport<>(TestApplication.class, resourceFilePath("grpc-test-config.yaml"));

    ManagedChannel channel = null;
    try {
        testSupport.before();
        channel = createPlaintextChannel(testSupport);
        final PersonServiceGrpc.PersonServiceBlockingStub client = PersonServiceGrpc.newBlockingStub(channel);

        testSupport.after();

        try {
            // this should fail as the server is now stopped
            client.getPerson(GetPersonRequest.newBuilder().setName("blah").build());
            fail("Request should have failed.");
        } catch (final Exception e) {
            assertEquals(StatusRuntimeException.class, e.getClass());
            assertEquals(Code.UNAVAILABLE, ((StatusRuntimeException) e).getStatus().getCode());
        }
    } finally {
        testSupport.after();
        shutdownChannel(channel);
    }
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test(timeout = 10000)
public void maxInboundSize_tooBig() {
    StreamingOutputCallRequest request =
            StreamingOutputCallRequest.newBuilder()
                                      .addResponseParameters(ResponseParameters.newBuilder().setSize(1))
                                      .build();
    int size = blockingStub.streamingOutputCall(request).next().getSerializedSize();

    TestServiceBlockingStub stub =
            Clients.newDerivedClient(
                    blockingStub,
                    GrpcClientOptions.MAX_INBOUND_MESSAGE_SIZE_BYTES.newValue(size - 1));
    Throwable t = catchThrowable(() -> stub.streamingOutputCall(request).next());
    assertThat(t).isInstanceOf(StatusRuntimeException.class);
    assertThat(((StatusRuntimeException) t).getStatus().getCode()).isEqualTo(Code.RESOURCE_EXHAUSTED);
    assertThat(Throwables.getStackTraceAsString(t)).contains("exceeds maximum");
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test(timeout = 10000)
public void maxOutboundSize_tooBig() {
    // set at least one field to ensure the size is non-zero.
    StreamingOutputCallRequest request =
            StreamingOutputCallRequest.newBuilder()
                                      .addResponseParameters(ResponseParameters.newBuilder().setSize(1))
                                      .build();
    TestServiceBlockingStub stub =
            Clients.newDerivedClient(
                    blockingStub,
                    GrpcClientOptions.MAX_OUTBOUND_MESSAGE_SIZE_BYTES.newValue(
                            request.getSerializedSize() - 1));
    Throwable t = catchThrowable(() -> stub.streamingOutputCall(request).next());
    assertThat(t).isInstanceOf(StatusRuntimeException.class);
    assertThat(((StatusRuntimeException) t).getStatus().getCode()).isEqualTo(Code.CANCELLED);
    assertThat(Throwables.getStackTraceAsString(t)).contains("message too large");
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void tooLargeRequest_uncompressed() throws Exception {
    SimpleRequest request = SimpleRequest.newBuilder()
                                         .setPayload(
                                                 Payload.newBuilder()
                                                        .setBody(ByteString.copyFrom(
                                                                LARGE_PAYLOAD.toByteArray())))
                                         .build();
    StatusRuntimeException t =
            (StatusRuntimeException) catchThrowable(
                    () -> blockingClient.staticUnaryCall(request));
    // NB: Since gRPC does not support HTTP/1, it just resets the stream with an HTTP/2 CANCEL error code,
    // which clients would interpret as Code.CANCELLED. Armeria supports HTTP/1, so more generically returns
    // an HTTP 500.
    assertThat(t.getStatus().getCode()).isEqualTo(Code.UNKNOWN);
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void tooLargeRequest_compressed() throws Exception {
    SimpleRequest request = SimpleRequest.newBuilder()
                                         .setPayload(
                                                 Payload.newBuilder()
                                                        .setBody(ByteString.copyFrom(
                                                                LARGE_PAYLOAD.toByteArray())))
                                         .build();
    StatusRuntimeException t =
            (StatusRuntimeException) catchThrowable(
                    () -> blockingClient.withCompression("gzip").staticUnaryCall(request));
    // NB: Since gRPC does not support HTTP/1, it just resets the stream with an HTTP/2 CANCEL error code,
    // which clients would interpret as Code.CANCELLED. Armeria supports HTTP/1, so more generically returns
    // an HTTP 500.
    assertThat(t.getStatus().getCode()).isEqualTo(Code.UNKNOWN);
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void maxMessageSizeShouldBeEnforced() throws Exception {
  // Allow the response payloads of up to 1 byte.
  startTransport(3, null, true, 1, null);

  MockStreamListener listener = new MockStreamListener();
  OkHttpClientStream stream =
      clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
  stream.start(listener);
  stream.request(1);
  assertContainStream(3);
  frameHandler().headers(false, false, 3, 0, grpcResponseHeaders(), HeadersMode.HTTP_20_HEADERS);
  assertNotNull(listener.headers);

  // Receive the message.
  final String message = "Hello Client";
  Buffer buffer = createMessageFrame(message);
  frameHandler().data(false, 3, buffer, (int) buffer.size());

  listener.waitUntilStreamClosed();
  assertEquals(Code.RESOURCE_EXHAUSTED, listener.status.getCode());
  shutdownAndVerify();
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void ping_failsIfTransportFails() throws Exception {
  initTransport();
  PingCallbackImpl callback = new PingCallbackImpl();
  clientTransport.ping(callback, MoreExecutors.directExecutor());
  assertEquals(1, clientTransport.getStats().get().keepAlivesSent);
  assertEquals(0, callback.invocationCount);

  clientTransport.onException(new IOException());
  // ping failed on error
  assertEquals(1, callback.invocationCount);
  assertTrue(callback.failureCause instanceof StatusException);
  assertEquals(Status.Code.UNAVAILABLE,
      ((StatusException) callback.failureCause).getStatus().getCode());

  // now that handler is in terminal state, all future pings fail immediately
  callback = new PingCallbackImpl();
  clientTransport.ping(callback, MoreExecutors.directExecutor());
  assertEquals(1, clientTransport.getStats().get().keepAlivesSent);
  assertEquals(1, callback.invocationCount);
  assertTrue(callback.failureCause instanceof StatusException);
  assertEquals(Status.Code.UNAVAILABLE,
      ((StatusException) callback.failureCause).getStatus().getCode());
  shutdownAndVerify();
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void streamErrorShouldNotCloseChannel() throws Exception {
  manualSetUp();
  createStream();
  stream.request(1);

  // When a DATA frame is read, throw an exception. It will be converted into an
  // Http2StreamException.
  RuntimeException e = new RuntimeException("Fake Exception");
  doThrow(e).when(streamListener).messagesAvailable(any(StreamListener.MessageProducer.class));

  // Read a DATA frame to trigger the exception.
  channelRead(emptyGrpcFrame(STREAM_ID, true));

  // Verify that the channel was NOT closed.
  assertTrue(channel().isOpen());

  // Verify the stream was closed.
  ArgumentCaptor<Status> captor = ArgumentCaptor.forClass(Status.class);
  verify(streamListener).closed(captor.capture());
  assertEquals(e, captor.getValue().asException().getCause());
  assertEquals(Code.UNKNOWN, captor.getValue().getCode());
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void headersWithInvalidContentTypeShouldFail() throws Exception {
  manualSetUp();
  Http2Headers headers = new DefaultHttp2Headers()
      .method(HTTP_METHOD)
      .set(CONTENT_TYPE_HEADER, new AsciiString("application/bad", UTF_8))
      .set(TE_HEADER, TE_TRAILERS)
      .path(new AsciiString("/foo/bar"));
  ByteBuf headersFrame = headersFrame(STREAM_ID, headers);
  channelRead(headersFrame);
  Http2Headers responseHeaders = new DefaultHttp2Headers()
      .set(InternalStatus.CODE_KEY.name(), String.valueOf(Code.INTERNAL.value()))
      .set(InternalStatus.MESSAGE_KEY.name(), "Content-Type 'application/bad' is not supported")
      .status("" + 415)
      .set(CONTENT_TYPE_HEADER, "text/plain; encoding=utf-8");

  verifyWrite().writeHeaders(eq(ctx()), eq(STREAM_ID), eq(responseHeaders), eq(0),
      eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(false), any(ChannelPromise.class));
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void headersWithInvalidMethodShouldFail() throws Exception {
  manualSetUp();
  Http2Headers headers = new DefaultHttp2Headers()
      .method(HTTP_FAKE_METHOD)
      .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC)
      .path(new AsciiString("/foo/bar"));
  ByteBuf headersFrame = headersFrame(STREAM_ID, headers);
  channelRead(headersFrame);
  Http2Headers responseHeaders = new DefaultHttp2Headers()
      .set(InternalStatus.CODE_KEY.name(), String.valueOf(Code.INTERNAL.value()))
      .set(InternalStatus.MESSAGE_KEY.name(), "Method 'FAKE' is not supported")
      .status("" + 405)
      .set(CONTENT_TYPE_HEADER, "text/plain; encoding=utf-8");

  verifyWrite().writeHeaders(eq(ctx()), eq(STREAM_ID), eq(responseHeaders), eq(0),
      eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(false), any(ChannelPromise.class));
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void headersWithMissingPathShouldFail() throws Exception {
  manualSetUp();
  Http2Headers headers = new DefaultHttp2Headers()
      .method(HTTP_METHOD)
      .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC);
  ByteBuf headersFrame = headersFrame(STREAM_ID, headers);
  channelRead(headersFrame);
  Http2Headers responseHeaders = new DefaultHttp2Headers()
      .set(InternalStatus.CODE_KEY.name(), String.valueOf(Code.UNIMPLEMENTED.value()))
      .set(InternalStatus.MESSAGE_KEY.name(), "Expected path but is missing")
      .status("" + 404)
      .set(CONTENT_TYPE_HEADER, "text/plain; encoding=utf-8");

  verifyWrite().writeHeaders(eq(ctx()), eq(STREAM_ID), eq(responseHeaders), eq(0),
      eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(false), any(ChannelPromise.class));
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void headersWithInvalidPathShouldFail() throws Exception {
  manualSetUp();
  Http2Headers headers = new DefaultHttp2Headers()
      .method(HTTP_METHOD)
      .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC)
      .path(new AsciiString("foo/bar"));
  ByteBuf headersFrame = headersFrame(STREAM_ID, headers);
  channelRead(headersFrame);
  Http2Headers responseHeaders = new DefaultHttp2Headers()
      .set(InternalStatus.CODE_KEY.name(), String.valueOf(Code.UNIMPLEMENTED.value()))
      .set(InternalStatus.MESSAGE_KEY.name(), "Expected path to start with /: foo/bar")
      .status("" + 404)
      .set(CONTENT_TYPE_HEADER, "text/plain; encoding=utf-8");

  verifyWrite().writeHeaders(eq(ctx()), eq(STREAM_ID), eq(responseHeaders), eq(0),
      eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(false), any(ChannelPromise.class));
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void maxMessageSizeShouldBeEnforced() throws Throwable {
  startServer();
  // Allow the response payloads of up to 1 byte.
  NettyClientTransport transport = newTransport(newNegotiator(),
      1, GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE, null, true);
  callMeMaybe(transport.start(clientTransportListener));

  try {
    // Send a single RPC and wait for the response.
    new Rpc(transport).halfClose().waitForResponse();
    fail("Expected the stream to fail.");
  } catch (ExecutionException e) {
    Status status = Status.fromThrowable(e);
    assertEquals(Code.RESOURCE_EXHAUSTED, status.getCode());
    assertTrue("Missing exceeds maximum from: " + status.getDescription(),
        status.getDescription().contains("exceeds maximum"));
  }
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void maxHeaderListSizeShouldBeEnforcedOnClient() throws Exception {
  startServer();

  NettyClientTransport transport =
      newTransport(newNegotiator(), DEFAULT_MAX_MESSAGE_SIZE, 1, null, true);
  callMeMaybe(transport.start(clientTransportListener));

  try {
    // Send a single RPC and wait for the response.
    new Rpc(transport, new Metadata()).halfClose().waitForResponse();
    fail("The stream should have been failed due to client received header exceeds header list"
        + " size limit!");
  } catch (Exception e) {
    Throwable rootCause = getRootCause(e);
    Status status = ((StatusException) rootCause).getStatus();
    assertEquals(Status.Code.INTERNAL, status.getCode());
    assertEquals("HTTP/2 error code: PROTOCOL_ERROR\nReceived Rst Stream",
        status.getDescription());
  }
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void maxHeaderListSizeShouldBeEnforcedOnServer() throws Exception {
  startServer(100, 1);

  NettyClientTransport transport = newTransport(newNegotiator());
  callMeMaybe(transport.start(clientTransportListener));

  try {
    // Send a single RPC and wait for the response.
    new Rpc(transport, new Metadata()).halfClose().waitForResponse();
    fail("The stream should have been failed due to server received header exceeds header list"
        + " size limit!");
  } catch (Exception e) {
    Status status = Status.fromThrowable(e);
    assertEquals(status.toString(), Status.Code.INTERNAL, status.getCode());
  }
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void cancel_doNotAcceptOk() {
  for (Code code : Code.values()) {
    ClientStreamListener listener = new NoopClientStreamListener();
    AbstractClientStream stream =
        new BaseAbstractClientStream(allocator, statsTraceCtx, transportTracer);
    stream.start(listener);
    if (code != Code.OK) {
      stream.cancel(Status.fromCodeValue(code.value()));
    } else {
      try {
        stream.cancel(Status.fromCodeValue(code.value()));
        fail();
      } catch (IllegalArgumentException e) {
        // ignore
      }
    }
  }
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void transportHeadersReceived_wrongContentType_401() {
  BaseTransportState state = new BaseTransportState(transportTracer);
  state.setListener(mockListener);
  Metadata headers = new Metadata();
  headers.put(testStatusMashaller, "401");
  headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), "text/html");
  state.transportHeadersReceived(headers);
  state.transportDataReceived(ReadableBuffers.empty(), true);

  verify(mockListener, never()).headersRead(any(Metadata.class));
  verify(mockListener).closed(statusCaptor.capture(), same(headers));
  assertEquals(Code.UNAUTHENTICATED, statusCaptor.getValue().getCode());
  assertTrue(statusCaptor.getValue().getDescription().contains("401"));
  assertTrue(statusCaptor.getValue().getDescription().contains("text/html"));
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void transportHeadersReceived_twice() {
  BaseTransportState state = new BaseTransportState(transportTracer);
  state.setListener(mockListener);
  Metadata headers = new Metadata();
  headers.put(testStatusMashaller, "200");
  headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER),
      "application/grpc");
  state.transportHeadersReceived(headers);
  Metadata headersAgain = new Metadata();
  state.transportHeadersReceived(headersAgain);
  state.transportDataReceived(ReadableBuffers.empty(), true);

  verify(mockListener).headersRead(headers);
  verify(mockListener).closed(statusCaptor.capture(), same(headersAgain));
  assertEquals(Code.INTERNAL, statusCaptor.getValue().getCode());
  assertTrue(statusCaptor.getValue().getDescription().contains("twice"));
}
 

import io.grpc.Status.Code; //导入依赖的package包/类
@Test
public void transportHeadersReceived_unknownAndTwiceLogsSecondHeaders() {
  BaseTransportState state = new BaseTransportState(transportTracer);
  state.setListener(mockListener);
  Metadata headers = new Metadata();
  headers.put(testStatusMashaller, "200");
  headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), "text/html");
  state.transportHeadersReceived(headers);
  Metadata headersAgain = new Metadata();
  String testString = "This is a test";
  headersAgain.put(Metadata.Key.of("key", Metadata.ASCII_STRING_MARSHALLER), testString);
  state.transportHeadersReceived(headersAgain);
  state.transportDataReceived(ReadableBuffers.empty(), true);

  verify(mockListener, never()).headersRead(any(Metadata.class));
  verify(mockListener).closed(statusCaptor.capture(), same(headers));
  assertEquals(Code.UNKNOWN, statusCaptor.getValue().getCode());
  assertTrue(statusCaptor.getValue().getDescription().contains(testString));
}