Java 类io.grpc.Grpc 实例源码

/** Helper for asserting TLS info in SSLSession {@link io.grpc.ServerCall#getAttributes()} */
protected void assertX500SubjectDn(String tlsInfo) {
  TestServiceGrpc.TestServiceBlockingStub stub =
      blockingStub.withDeadlineAfter(5, TimeUnit.SECONDS);

  stub.unaryCall(SimpleRequest.getDefaultInstance());

  List<Certificate> certificates = Lists.newArrayList();
  SSLSession sslSession =
      serverCallCapture.get().getAttributes().get(Grpc.TRANSPORT_ATTR_SSL_SESSION);
  try {
    certificates = Arrays.asList(sslSession.getPeerCertificates());
  } catch (SSLPeerUnverifiedException e) {
    // Should never happen
    throw new AssertionError(e);
  }

  X509Certificate x509cert = (X509Certificate) certificates.get(0);

  assertEquals(1, certificates.size());
  assertEquals(tlsInfo, x509cert.getSubjectDN().toString());
}

/**
 * Create a server plaintext handler for gRPC.
 */
public static ProtocolNegotiator serverPlaintext() {
  return new ProtocolNegotiator() {
    @Override
    public Handler newHandler(final GrpcHttp2ConnectionHandler handler) {
      class PlaintextHandler extends ChannelHandlerAdapter implements Handler {
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
          // Set sttributes before replace to be sure we pass it before accepting any requests.
          handler.handleProtocolNegotiationCompleted(Attributes.newBuilder()
              .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, ctx.channel().remoteAddress())
              .build());
          // Just replace this handler with the gRPC handler.
          ctx.pipeline().replace(this, null, handler);
        }

        @Override
        public AsciiString scheme() {
          return Utils.HTTP;
        }
      }

      return new PlaintextHandler();
    }
  };
}

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  if (evt instanceof SslHandshakeCompletionEvent) {
    SslHandshakeCompletionEvent handshakeEvent = (SslHandshakeCompletionEvent) evt;
    if (handshakeEvent.isSuccess()) {
      if (NEXT_PROTOCOL_VERSIONS.contains(sslHandler(ctx.pipeline()).applicationProtocol())) {
        // Successfully negotiated the protocol.
        // Notify about completion and pass down SSLSession in attributes.
        grpcHandler.handleProtocolNegotiationCompleted(
            Attributes.newBuilder()
                .set(Grpc.TRANSPORT_ATTR_SSL_SESSION,
                    sslHandler(ctx.pipeline()).engine().getSession())
                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, ctx.channel().remoteAddress())
                .build());
        // Replace this handler with the GRPC handler.
        ctx.pipeline().replace(this, null, grpcHandler);
      } else {
        fail(ctx, new Exception(
            "Failed protocol negotiation: Unable to find compatible protocol."));
      }
    } else {
      fail(ctx, handshakeEvent.cause());
    }
  }
  super.userEventTriggered(ctx, evt);
}

@Override
public void onClose(Status status, Metadata trailers) {
  try {
    SocketAddress remoteServer = clientCall.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
    callOptions.getOption(GrpcCallOptions.CALLOPTIONS_CUSTOME_KEY)
        .put(GrpcCallOptions.GRPC_CURRENT_ADDR_KEY, remoteServer);
  } finally {
    if (status.isOk()) {
      statusOk(trailers);
    } else {
      statusError(status, trailers);
    }
  }
}

/** Helper for getting remote address {@link io.grpc.ServerCall#getAttributes()} */
protected SocketAddress obtainRemoteClientAddr() {
  TestServiceGrpc.TestServiceBlockingStub stub =
      blockingStub.withDeadlineAfter(5, TimeUnit.SECONDS);

  stub.unaryCall(SimpleRequest.getDefaultInstance());

  return serverCallCapture.get().getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
}

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  if (evt instanceof SslHandshakeCompletionEvent) {
    SslHandshakeCompletionEvent handshakeEvent = (SslHandshakeCompletionEvent) evt;
    if (handshakeEvent.isSuccess()) {
      SslHandler handler = ctx.pipeline().get(SslHandler.class);
      if (NEXT_PROTOCOL_VERSIONS.contains(handler.applicationProtocol())) {
        // Successfully negotiated the protocol.
        logSslEngineDetails(Level.FINER, ctx, "TLS negotiation succeeded.", null);

        // Wait until negotiation is complete to add gRPC.   If added too early, HTTP/2 writes
        // will fail before we see the userEvent, and the channel is closed down prematurely.
        ctx.pipeline().addBefore(ctx.name(), null, grpcHandler);

        // Successfully negotiated the protocol.
        // Notify about completion and pass down SSLSession in attributes.
        grpcHandler.handleProtocolNegotiationCompleted(
            Attributes.newBuilder()
                .set(Grpc.TRANSPORT_ATTR_SSL_SESSION, handler.engine().getSession())
                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, ctx.channel().remoteAddress())
                .build());
        writeBufferedAndRemove(ctx);
      } else {
        Exception ex = new Exception(
            "Failed ALPN negotiation: Unable to find compatible protocol.");
        logSslEngineDetails(Level.FINE, ctx, "TLS negotiation failed.", ex);
        fail(ctx, ex);
      }
    } else {
      fail(ctx, handshakeEvent.cause());
    }
  }
  super.userEventTriggered(ctx, evt);
}

@Test
public void clientStreamGetsAttributes() throws Exception {
  startServer();
  NettyClientTransport transport = newTransport(newNegotiator());
  callMeMaybe(transport.start(clientTransportListener));
  Rpc rpc = new Rpc(transport).halfClose();
  rpc.waitForResponse();

  assertNotNull(rpc.stream.getAttributes().get(Grpc.TRANSPORT_ATTR_SSL_SESSION));
  assertEquals(address, rpc.stream.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR));
}

@CheckReturnValue
@Override
public synchronized Runnable start(ManagedClientTransport.Listener listener) {
  this.clientTransportListener = listener;
  InProcessServer server = InProcessServer.findServer(name);
  if (server != null) {
    serverSchedulerPool = server.getScheduledExecutorServicePool();
    serverScheduler = serverSchedulerPool.getObject();
    serverStreamTracerFactories = server.getStreamTracerFactories();
    // Must be semi-initialized; past this point, can begin receiving requests
    serverTransportListener = server.register(this);
  }
  if (serverTransportListener == null) {
    shutdownStatus = Status.UNAVAILABLE.withDescription("Could not find server: " + name);
    final Status localShutdownStatus = shutdownStatus;
    return new Runnable() {
      @Override
      public void run() {
        synchronized (InProcessTransport.this) {
          notifyShutdown(localShutdownStatus);
          notifyTerminated();
        }
      }
    };
  }
  return new Runnable() {
    @Override
    @SuppressWarnings("deprecation")
    public void run() {
      synchronized (InProcessTransport.this) {
        Attributes serverTransportAttrs = Attributes.newBuilder()
            .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InProcessSocketAddress(name))
            .build();
        serverStreamAttributes = serverTransportListener.transportReady(serverTransportAttrs);
        clientTransportListener.transportReady();
      }
    }
  };
}

private void contextCopy(ServerCall<?, ?> call, final Metadata headers) {
  copyMetadataToThreadLocal(headers);
  InetSocketAddress remoteAddress =
      (InetSocketAddress) call.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
  RpcContext.getContext().setAttachment(Constants.REMOTE_ADDRESS, remoteAddress.getHostString());
}