Java 类io.grpc.netty.NegotiationType 实例源码

public void connectPlugin(String host, int port) {
    ManagedChannel channel = NettyChannelBuilder.forAddress(host, port)
            .negotiationType(NegotiationType.PLAINTEXT) // TODO: gRPC encryption
            .keepAliveTime(1, TimeUnit.MINUTES)
            .keepAliveTimeout(5, TimeUnit.SECONDS)
            .directExecutor()
            .channelType(EpollSocketChannel.class)
            .eventLoopGroup(new EpollEventLoopGroup())
            .build();

    PluginManagerGrpc.PluginManagerBlockingStub blocking = PluginManagerGrpc.newBlockingStub(channel);
    PluginManagerGrpc.PluginManagerStub async = PluginManagerGrpc.newStub(channel);

    ServiceConnection connection = ServiceConnection.builder()
            .channel(channel)
            .blockingStub(blocking)
            .asyncStub(async)
            .build();

    this.pluginConnections.put(PLUGIN_MANAGER, connection);
}

public MemberServiceImpl(String host, int port, Crypto crypto) {
    Preconditions.checkNotNull(host);
    Preconditions.checkNotNull(port);

    InetAddress address = null;
    try {
        address = InetAddress.getByName(host);
    } catch (UnknownHostException e) {
        logger.error("Create member service failed by unknown host exception", e);
        Throwables.propagate(e);
    }

    final Channel channel = NettyChannelBuilder
            .forAddress(new InetSocketAddress(address, port))
            .negotiationType(NegotiationType.PLAINTEXT)
            .build();

    initializeStubs(channel);
    this.crypto = crypto;
}

@Override
public PubsubClient newClient(
    @Nullable String timestampAttribute, @Nullable String idAttribute, PubsubOptions options)
    throws IOException {
  ManagedChannel channel = NettyChannelBuilder
      .forAddress(PUBSUB_ADDRESS, PUBSUB_PORT)
      .negotiationType(NegotiationType.TLS)
      .sslContext(GrpcSslContexts.forClient().ciphers(null).build())
      .build();

  return new PubsubGrpcClient(timestampAttribute,
                              idAttribute,
                              DEFAULT_TIMEOUT_S,
                              channel,
                              options.getGcpCredential());
}

private static ManagedChannel getGenomicsManagedChannel(List<ClientInterceptor> interceptors)
    throws SSLException {
  // Java 8's implementation of GCM ciphers is extremely slow. Therefore we disable
  // them here.
  List<String> defaultCiphers = GrpcSslContexts.forClient().ciphers(null).build().cipherSuites();
  List<String> performantCiphers = new ArrayList<>();
  for (String cipher : defaultCiphers) {
    if (!cipher.contains("GCM")) {
      performantCiphers.add(cipher);
    }
  }

  return NettyChannelBuilder.forAddress(GENOMICS_ENDPOINT, 443)
      .negotiationType(NegotiationType.TLS)
      .sslContext(GrpcSslContexts.forClient().ciphers(performantCiphers).build())
      .intercept(interceptors)
      .build();
}

private ManagedChannel newClientChannel() throws CertificateException, IOException {
  File clientCertChainFile = TestUtils.loadCert("client.pem");
  File clientPrivateKeyFile = TestUtils.loadCert("client.key");
  X509Certificate[] clientTrustedCaCerts = {
    TestUtils.loadX509Cert("ca.pem")
  };

  SslContext sslContext =
      GrpcSslContexts.forClient()
                     .keyManager(clientCertChainFile, clientPrivateKeyFile)
                     .trustManager(clientTrustedCaCerts)
                     .build();

  return NettyChannelBuilder.forAddress("localhost", server.getPort())
      .overrideAuthority(TestUtils.TEST_SERVER_HOST)
      .negotiationType(NegotiationType.TLS)
      .sslContext(sslContext)
      .build();
}

/**
 * Constructor for the client.
 * @param host The host to connect to for the server
 * @param port The port to connect to on the host server
 */
public MyServiceClient(String host, int port) {
  InetAddress address;
  try {
    address = InetAddress.getByName(host);
  } catch (UnknownHostException e) {
    throw new RuntimeException(e);
  }
  managedChannel = NettyChannelBuilder.forAddress(new InetSocketAddress(address,port))
      .flowControlWindow(65 * 1024)
      .negotiationType(NegotiationType.PLAINTEXT).build();
  simpleBlockingStub = SimpleGrpc.newBlockingStub(managedChannel);
  lessSimpleBlockingStub = LessSimpleGrpc.newBlockingStub(managedChannel);
}

private static LoadingCache<UserAuth, String> initCache() {
  try {
    currentAuth = ApprcHolder.get().currentAuth();
    URI uri = new URI(currentAuth.getApiserver());
    logger.info(String.format("Connecting to apiserver: %s  host: %s  port: %s",
        currentAuth.getApiserver(), uri.getHost(), uri.getPort()));
    NettyChannelBuilder builder = NettyChannelBuilder
        .forAddress(uri.getHost(), uri.getPort())
        .nameResolverFactory(new DnsNameResolverProvider());
    if (useTLS(currentAuth)) {
      File trustCertCollectionFile = null;
      builder
          .sslContext(GrpcSslContexts.forClient().trustManager(trustCertCollectionFile).build())
          .negotiationType(NegotiationType.TLS);
    } else {
      builder.negotiationType(NegotiationType.PLAINTEXT);
    }
    channel = builder.build();
    return CacheBuilder.newBuilder()
        .expireAfterAccess(DESCRIPTOR.getAuthCacheTtl(), TimeUnit.SECONDS)
        .build(
            new CacheLoader<UserAuth, String>() {
              @Override
              public String load(UserAuth key) throws Exception {
                if (isToken(key.getSecret())) {
                  return checkToken(key.getUsername(),
                      key.getSecret().substring(BEARER_PREFIX.length()));
                }
                return checkPassword(key.getUsername(), key.getSecret());
              }
            }
        );
  } catch (URISyntaxException | SSLException e) {
    logger.log(Level.SEVERE, e.getMessage());
  }
  return null;
}

@SuppressWarnings("unchecked")
@Override
public void addConnection(ServiceType serviceType, String host, int port) {
    ManagedChannel channel = NettyChannelBuilder.forAddress(host, port)
            .negotiationType(NegotiationType.PLAINTEXT) // TODO: gRPC encryption
            .keepAliveTime(1, TimeUnit.MINUTES)
            .keepAliveTimeout(5, TimeUnit.SECONDS)
            .directExecutor()
            .channelType(EpollSocketChannel.class)
            .eventLoopGroup(new EpollEventLoopGroup())
            .build();

    AbstractStub blocking;
    AbstractStub async;

    switch (serviceType) {
        case WORLD: {
            blocking = WorldServiceGrpc.newBlockingStub(channel);
            async = WorldServiceGrpc.newStub(channel);
            break;
        }
        case PLUGIN_MANAGER: {
            blocking = PluginManagerGrpc.newBlockingStub(channel);
            async = PluginManagerGrpc.newStub(channel);
            break;
        }
        default: {
            throw new RuntimeException("Service type not handled: " + serviceType.name());
        }
    }

    ServiceConnection connection = ServiceConnection.builder()
            .channel(channel)
            .blockingStub(blocking)
            .asyncStub(async)
            .build();

    this.connections.put(serviceType, connection);
}

private ManagedChannel createChannel(URI uri) {
    log.debug("Creating channel for {}", uri);
    int port = GrpcRemoteServiceServer.DEFAULT_LISTEN_PORT;
    if (uri.getPort() != -1) {
        port = uri.getPort();
    }
    return NettyChannelBuilder.forAddress(uri.getHost(), port)
            .negotiationType(NegotiationType.PLAINTEXT)
            .build();
}

private NettyChannelBuilder createChannelBuilder(HostAndPort endpoint) {
  if (!callConfiguration.getUseTls()) {
    return NettyChannelBuilder.forAddress(endpoint.getHostText(), endpoint.getPort())
        .negotiationType(NegotiationType.PLAINTEXT);
  } else {
    return NettyChannelBuilder.forAddress(endpoint.getHostText(), endpoint.getPort())
        .sslContext(createSslContext())
        .negotiationType(NegotiationType.TLS)
        .intercept(metadataInterceptor());
  }
}

public GRPCClient(String host, int port, int observerPort) {
    log.debug("Trying to connect to GRPC host:port={}:{}, host:observerPort={}:{}, ", host, port, observerPort);
    ManagedChannel channel = NettyChannelBuilder.forAddress(host, port).negotiationType(NegotiationType.PLAINTEXT).build();
    ManagedChannel observerChannel = NettyChannelBuilder.forAddress(host, observerPort).negotiationType(NegotiationType.PLAINTEXT).build();
    pbs = PeerGrpc.newBlockingStub(channel);
    obs = OpenchainGrpc.newBlockingStub(channel);
    observer = new GRPCObserver(observerChannel);
    observer.connect();
}

/** Return {@link io.grpc.Channel} which is used by Cloud Pub/Sub gRPC API's. */
public static Channel getChannel() throws IOException {
  ManagedChannel channelImpl =
      NettyChannelBuilder.forAddress(ENDPOINT, 443).negotiationType(NegotiationType.TLS).build();
  final ClientAuthInterceptor interceptor =
      new ClientAuthInterceptor(
          GoogleCredentials.getApplicationDefault().createScoped(CPS_SCOPE),
          Executors.newCachedThreadPool());
  return ClientInterceptors.intercept(channelImpl, interceptor);
}

public GRPCClient(String host, int port, int observerPort) {
    log.debug("Trying to connect to GRPC host:port={}:{}, host:observerPort={}:{}, ", host, port, observerPort);
    ManagedChannel channel = NettyChannelBuilder.forAddress(host, port).negotiationType(NegotiationType.PLAINTEXT).build();
    ManagedChannel observerChannel = NettyChannelBuilder.forAddress(host, observerPort).negotiationType(NegotiationType.PLAINTEXT).build();
    dbs = DevopsGrpc.newBlockingStub(channel);
    obs = OpenchainGrpc.newBlockingStub(channel);
    observer = new GRPCObserver(observerChannel);
    observer.connect();
}

public static void main(final String[] args) throws Exception {

        if (args.length == 0) {
            System.err.println("Please specify your project name.");
            System.exit(1);
        }
        final String project = args[0];
        ManagedChannelImpl channelImpl = NettyChannelBuilder
            .forAddress("pubsub.googleapis.com", 443)
            .negotiationType(NegotiationType.TLS)
            .build();
        GoogleCredentials creds = GoogleCredentials.getApplicationDefault();
        // Down-scope the credential to just the scopes required by the service
        creds = creds.createScoped(Arrays.asList("https://www.googleapis.com/auth/pubsub"));
        // Intercept the channel to bind the credential
        ExecutorService executor = Executors.newSingleThreadExecutor();
        ClientAuthInterceptor interceptor = new ClientAuthInterceptor(creds, executor);
        Channel channel = ClientInterceptors.intercept(channelImpl, interceptor);
        // Create a stub using the channel that has the bound credential
        PublisherGrpc.PublisherBlockingStub publisherStub = PublisherGrpc.newBlockingStub(channel);
        ListTopicsRequest request = ListTopicsRequest.newBuilder()
                .setPageSize(10)
                .setProject("projects/" + project)
                .build();
        ListTopicsResponse resp = publisherStub.listTopics(request);
        System.out.println("Found " + resp.getTopicsCount() + " topics.");
        for (Topic topic : resp.getTopicsList()) {
            System.out.println(topic.getName());
        }
    }

/**
 * Create a new gRPC {@link ManagedChannel}.
 *
 * @throws IOException in case the channel can't be constructed.
 */
public static ManagedChannel newChannel(String target, AuthAndTLSOptions options)
    throws IOException {
  Preconditions.checkNotNull(target);
  Preconditions.checkNotNull(options);

  final SslContext sslContext =
      options.tlsEnabled ? createSSlContext(options.tlsCertificate) : null;

  try {
    NettyChannelBuilder builder =
        NettyChannelBuilder.forTarget(target)
            .negotiationType(options.tlsEnabled ? NegotiationType.TLS : NegotiationType.PLAINTEXT)
            .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance());
    if (sslContext != null) {
      builder.sslContext(sslContext);
      if (options.tlsAuthorityOverride != null) {
        builder.overrideAuthority(options.tlsAuthorityOverride);
      }
    }
    return builder.build();
  } catch (RuntimeException e) {
    // gRPC might throw all kinds of RuntimeExceptions: StatusRuntimeException,
    // IllegalStateException, NullPointerException, ...
    String message = "Failed to connect to '%s': %s";
    throw new IOException(String.format(message, target, e.getMessage()));
  }
}

private ManagedChannel createChannel(URI uri) {
    log.debug("Creating channel for {}", uri);
    int port = GrpcRemoteServiceServer.DEFAULT_LISTEN_PORT;
    if (uri.getPort() != -1) {
        port = uri.getPort();
    }
    return NettyChannelBuilder.forAddress(uri.getHost(), port)
            .negotiationType(NegotiationType.PLAINTEXT)
            // TODO Not ideal fix, gRPC discovers name resolvers
            // in the class path, but OSGi was preventing it.
            // Manually specifying the default dns resolver for now.
            .nameResolverFactory(new DnsNameResolverProvider())
            .build();
}

private void runTest() throws Exception {
  try {
    controlChannel = NettyChannelBuilder.forAddress("127.0.0.1", serverControlPort)
        .negotiationType(NegotiationType.PLAINTEXT).build();
    controlStub = ReconnectServiceGrpc.newBlockingStub(controlChannel);
    if (useOkhttp) {
      retryChannel = OkHttpChannelBuilder.forAddress("127.0.0.1", serverRetryPort)
          .negotiationType(io.grpc.okhttp.NegotiationType.TLS).build();
    } else {
      retryChannel = NettyChannelBuilder.forAddress("127.0.0.1", serverRetryPort)
          .negotiationType(NegotiationType.TLS).build();
    }
    retryStub = ReconnectServiceGrpc.newBlockingStub(retryChannel);
    controlStub.start(Empty.getDefaultInstance());

    long startTimeStamp = System.currentTimeMillis();
    while ((System.currentTimeMillis() - startTimeStamp) < TEST_TIME_MS) {
      try {
        retryStub.start(Empty.getDefaultInstance());
      } catch (StatusRuntimeException expected) {
        // Make CheckStyle happy.
      }
      Thread.sleep(50);
    }
    ReconnectInfo info = controlStub.stop(Empty.getDefaultInstance());
    assertTrue(info.getPassed());
  } finally {
    controlChannel.shutdownNow();
    retryChannel.shutdownNow();
  }
}

private ManagedChannel createChannel() {
  InetAddress address;
  try {
    address = InetAddress.getByName(serverHost);
  } catch (UnknownHostException ex) {
    throw new RuntimeException(ex);
  }
  return NettyChannelBuilder.forAddress(new InetSocketAddress(address, serverPort))
      .negotiationType(NegotiationType.PLAINTEXT)
      .build();
}

private ManagedChannel createChannel(InetSocketAddress address) {
  SslContext sslContext = null;
  if (useTestCa) {
    try {
      sslContext = GrpcSslContexts.forClient().trustManager(
          TestUtils.loadCert("ca.pem")).build();
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }
  return NettyChannelBuilder.forAddress(address)
      .negotiationType(useTls ? NegotiationType.TLS : NegotiationType.PLAINTEXT)
      .sslContext(sslContext)
      .build();
}

@Override
protected ManagedChannel createChannel() {
  NettyChannelBuilder builder = NettyChannelBuilder.forAddress("localhost", getPort())
      .negotiationType(NegotiationType.PLAINTEXT)
      .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE);
  io.grpc.internal.TestingAccessor.setStatsImplementation(
      builder, createClientCensusStatsModule());
  return builder.build();
}

private static ManagedChannel clientChannel(int port, SslContext sslContext) throws IOException {
  return NettyChannelBuilder.forAddress("localhost", port)
      .overrideAuthority(TestUtils.TEST_SERVER_HOST)
      .negotiationType(NegotiationType.TLS)
      .sslContext(sslContext)
      .build();
}

/**
 * Resets client/server and their flow control windows.
 */
private void resetConnection(int clientFlowControlWindow)
    throws InterruptedException {
  if (channel != null) {
    if (!channel.isShutdown()) {
      channel.shutdown();
      channel.awaitTermination(100, TimeUnit.MILLISECONDS);
    }
  }
  channel = NettyChannelBuilder.forAddress(new InetSocketAddress("localhost", proxyPort))
      .flowControlWindow(clientFlowControlWindow)
      .negotiationType(NegotiationType.PLAINTEXT)
      .build();
}

@Override
protected ManagedChannel createChannel() {
  NettyChannelBuilder builder = NettyChannelBuilder
      .forAddress(new LocalAddress("in-process-1"))
      .negotiationType(NegotiationType.PLAINTEXT)
      .channelType(LocalChannel.class)
      .flowControlWindow(65 * 1024)
      .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE);
  io.grpc.internal.TestingAccessor.setStatsImplementation(
      builder, createClientCensusStatsModule());
  return builder.build();
}

private static OkHttpChannelBuilder newOkhttpClientChannel(
    SocketAddress address, boolean tls, boolean testca, @Nullable String authorityOverride) {
  InetSocketAddress addr = (InetSocketAddress) address;
  OkHttpChannelBuilder builder =
      OkHttpChannelBuilder.forAddress(addr.getHostName(), addr.getPort());
  if (tls) {
    builder.negotiationType(io.grpc.okhttp.NegotiationType.TLS);
    SSLSocketFactory factory;
    if (testca) {
      builder.overrideAuthority(
          GrpcUtil.authorityFromHostAndPort(authorityOverride, addr.getPort()));
      try {
        factory = TestUtils.newSslSocketFactoryForCa(
            Platform.get().getProvider(),
            TestUtils.loadCert("ca.pem"));
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    } else {
      factory = (SSLSocketFactory) SSLSocketFactory.getDefault();
    }
    builder.sslSocketFactory(factory);
  } else {
    builder.negotiationType(io.grpc.okhttp.NegotiationType.PLAINTEXT);
  }
  return builder;
}

private ManagedChannel getNewManagedChannel() throws UnknownHostException {
    InetAddress address = InetAddress.getByName(host);
    return NettyChannelBuilder.forAddress(
            new InetSocketAddress(address, port)).flowControlWindow(65 * 1024)
            .negotiationType(NegotiationType.PLAINTEXT).build();
}

public Builder withNegotiationType(NegotiationType negotiationType) {
    this.negotiationType = negotiationType;
    return this;
}

private static ManagedChannel createChannel(String target) {
  NettyChannelBuilder builder =
      NettyChannelBuilder.forTarget(target)
          .negotiationType(NegotiationType.PLAINTEXT);
  return builder.build();
}

public static void main(String[] args) throws InterruptedException {
//        ManagedChannel channel = ManagedChannelBuilder.forAddress("172.16.101.200", 44134).usePlaintext(true).build();

        ManagedChannel channel = NettyChannelBuilder
                .forAddress("172.16.80.151", 44134)
                .negotiationType(NegotiationType.PLAINTEXT).build();

        ReleaseServiceGrpc.ReleaseServiceBlockingStub blockingStub = ReleaseServiceGrpc
                .newBlockingStub(channel).withCallCredentials(new CallCredentials(){

                    @Override
                    public void applyRequestMetadata(MethodDescriptor<?, ?> method, Attributes attrs,
                                                     Executor appExecutor, MetadataApplier applier) {
                        // TODO Auto-generated method stub
                        Metadata metadata = new Metadata();
                        metadata.put(Metadata.Key.of("x-helm-api-client", Metadata.ASCII_STRING_MARSHALLER), "v2.2.0");
                        applier.apply(metadata);
                    }

                });

        //deployAChart(blockingStub);
        //updateDeploy(blockingStub);
        //undeploy(blockingStub);
        Tiller.GetVersionResponse response = blockingStub.getVersion(Tiller.GetVersionRequest.newBuilder().build());
        System.out.println(response.getVersion());

        Iterator<Tiller.ListReleasesResponse> listReleasesIterator = blockingStub.listReleases(Tiller.ListReleasesRequest.newBuilder().build());
        System.out.println("================start=================");
        while (listReleasesIterator.hasNext()) {
            System.out.println(listReleasesIterator.next());
            System.out.println("---------------------------------");
        }
        System.out.println("================end=================");

        Tiller.GetReleaseStatusResponse releaseStatus = blockingStub.getReleaseStatus(Tiller.GetReleaseStatusRequest
                .newBuilder()
                .setName("invited-catfish").build());
        System.out.println("status:"+releaseStatus);

        System.out.println("-----------------");

        Tiller.GetHistoryResponse history = blockingStub.getHistory(Tiller.GetHistoryRequest.newBuilder()
                .setName("invited-catfish").setMax(10).build());
        System.out.println(history);

        try {
            channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

public static void main(String[] args) throws InterruptedException {
//        ManagedChannel channel = ManagedChannelBuilder.forAddress("172.16.101.200", 44134).usePlaintext(true).build();

        ManagedChannel channel = NettyChannelBuilder
                .forAddress("172.16.101.200", 44134)
                .negotiationType(NegotiationType.PLAINTEXT).build();

        ReleaseServiceGrpc.ReleaseServiceBlockingStub blockingStub = ReleaseServiceGrpc
                .newBlockingStub(channel).withCallCredentials(new CallCredentials(){

                    @Override
                    public void applyRequestMetadata(MethodDescriptor<?, ?> method, Attributes attrs,
                            Executor appExecutor, MetadataApplier applier) {
                        // TODO Auto-generated method stub
                          Metadata metadata = new Metadata();
                          metadata.put(Metadata.Key.of("x-helm-api-client", Metadata.ASCII_STRING_MARSHALLER), "v2.2.0");
                          applier.apply(metadata);
                    }

                });


        deployAChart(blockingStub);
        //updateDeploy(blockingStub);
        //undeploy(blockingStub);
       /* Tiller.GetVersionResponse response = blockingStub.getVersion(Tiller.GetVersionRequest.newBuilder().build());
        System.out.println(response.getVersion());

        Iterator<Tiller.ListReleasesResponse> listReleasesIterator = blockingStub.listReleases(Tiller.ListReleasesRequest.newBuilder().build());
        System.out.println("================start=================");
        while (listReleasesIterator.hasNext()) {
            System.out.println(listReleasesIterator.next());
            System.out.println("---------------------------------");
        }
        System.out.println("================end=================");

        Tiller.GetReleaseStatusResponse releaseStatus = blockingStub.getReleaseStatus(Tiller.GetReleaseStatusRequest
                .newBuilder()
                .setName("invited-catfish").build());
        System.out.println("status:"+releaseStatus);

        System.out.println("-----------------");

        Tiller.GetHistoryResponse history = blockingStub.getHistory(Tiller.GetHistoryRequest.newBuilder()
                .setName("invited-catfish").setMax(10).build());
        System.out.println(history);
*/
        try {
            channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

/**
 * Build an AsyncHandler instance
 *
 * @param _credentials A valid authentication token
 * @param _host The handler host
 * @param _port The handler port
 * @param _certificate The handler certificate
 * @return An Observable stream containing the newly built AsyncHandler wrapper
 */
public static Observable<AsyncHandler> from(AsyncOAuth2Token _credentials, String _host, int _port, InputStream _certificate) {

    return Observable
            .create((Subscriber<? super AsyncHandler> t) -> {
                try {
                    t.onNext(new AsyncHandler(
                            ApplicationManagerGrpc.newFutureStub(
                                    NettyChannelBuilder
                                    .forAddress(_host, _port)
                                    .negotiationType(NegotiationType.TLS)
                                    .sslContext(GrpcSslContexts
                                            .forClient()
                                            .trustManager(_certificate)
                                            .build()
                                    )
                                    .intercept(new ClientInterceptor() {
                                        @Override
                                        public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
                                            return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {

                                                @Override
                                                public void start(ClientCall.Listener<RespT> responseListener, Metadata headers) {
                                                    /**
                                                     * Add auth header here
                                                     */
                                                    headers.put(Metadata.Key.of("token", Metadata.ASCII_STRING_MARSHALLER), _credentials.getRawToken());
                                                    super.start(responseListener, headers);
                                                }
                                            };
                                        }
                                    })
                                    .build()
                            )
                    ));
                    t.onCompleted();
                } catch (Exception ex) {
                    t.onError(ex);
                }
            });

}

/**
 * <p>
 * createNettyChannel.
 * </p>
 *
 * @param interceptors a {@link List} object.
 * @param globalClientOptions a {@link ClientOptions} object.
 * @return a {@link ManagedChannel} object.
 * @throws SSLException if any.
 * @throws IOException if any.
 */
private ManagedChannel createNettyChannel(ClientOptions clientOptions, List<ClientInterceptor> interceptors) throws SSLException, IOException {

  NettyChannelBuilder builder = NettyChannelBuilder.forTarget(clientOptions.getServiceEndpoint());
  // 这里要注意下由于java版本的没有提供像go那样的可以指定域名
  // java版本源代码中把host传入作为证书域名
  // 域名是在证书生成的过程中自己输入的
  //String serverHostOverride = "localhost";
  if (clientOptions.getServerHostOverride() != null) {
    // Force the hostname to match the cert the server uses.
    builder.overrideAuthority(clientOptions.getServerHostOverride());
  }
  if (clientOptions.isUsedTls()) {
    builder
        .sslContext(
            GrpcSslContexts.forClient().trustManager(SslUtils.loadCert("server.pem")).build())
        .negotiationType(NegotiationType.TLS);
  }   
  builder
      .nameResolverFactory(AbstractNameResolverProvider.provider()
              .withRegistryCenterAddress(clientOptions.getRegistryCenterAddress())
              .withZoneToPrefer(clientOptions.getZoneToPrefer())
              .withServiceEndpoint(clientOptions.getServiceEndpoint())
              .withUsedTls(clientOptions.isUsedTls()))
      .loadBalancerFactory(GrpclbLoadBalancerFactory.getInstance())
      //The TCP connections are shutdown when you shutdown the Channel. 
      //Specify an idleTimeout() to have the Channel automatically close the TCP connection after a period of inactivity.
      .idleTimeout(Long.MAX_VALUE, TimeUnit.SECONDS)
      .maxInboundMessageSize(MAX_MESSAGE_SIZE)
      //.sslContext(createSslContext())
      //TODO: Caused run unit testing error happen in maven if comment out below 1 line code!!!
      //.eventLoopGroup(RpcSharedThreadPools.getInstance().getElg())
      .executor(RpcSharedThreadPools.getInstance().getBatchThreadPool())
      // .userAgent(VersionInfo.CORE_UESR_AGENT + "," + options.getUserAgent())
      .flowControlWindow(FLOW_CONTROL_WINDOW)
      .intercept(new HeaderClientInterceptor(),
          new StubDeadlineClientInterceptor());
  if(interceptors != null && interceptors.size() > 0){
    builder.intercept(interceptors);
  }
  return builder.build();
}

private Channel createGrpcChannel() {
  return NettyChannelBuilder.forAddress("localhost", grpcPort)
      .negotiationType(NegotiationType.PLAINTEXT)
      .build();
}

CentralConnection(String collectorAddress, @Nullable String collectorAuthority, File confDir,
        @Nullable File sharedConfDir, AtomicBoolean inConnectionFailure) throws SSLException {
    ParsedCollectorAddress parsedCollectorAddress = parseCollectorAddress(collectorAddress);
    eventLoopGroup = EventLoopGroups.create("Glowroot-GRPC-Worker-ELG");
    channelExecutor =
            Executors.newSingleThreadExecutor(ThreadFactories.create("Glowroot-GRPC-Executor"));
    String authority;
    if (collectorAuthority != null) {
        authority = collectorAuthority;
    } else if (parsedCollectorAddress.addresses().size() == 1) {
        authority = parsedCollectorAddress.addresses().get(0).getHostName();
    } else if (!parsedCollectorAddress.https()) {
        authority = "dummy-service-authority";
    } else {
        throw new IllegalStateException("collector.authority is required when using client"
                + " side load balancing to connect to a glowroot central cluster over HTTPS");
    }
    NettyChannelBuilder builder = NettyChannelBuilder
            .forTarget("dummy-target")
            .nameResolverFactory(new SimpleNameResolverFactory(
                    parsedCollectorAddress.addresses(), authority))
            .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance())
            .eventLoopGroup(eventLoopGroup)
            .executor(channelExecutor)
            // aggressive keep alive, shouldn't even be used since gauge data is sent every
            // 5 seconds and keep alive will only kick in after 30 seconds of not hearing back
            // from the server
            .keepAliveTime(30, SECONDS);
    if (parsedCollectorAddress.https()) {
        SslContextBuilder sslContext = GrpcSslContexts.forClient();
        File trustCertCollectionFile = getTrustCertCollectionFile(confDir, sharedConfDir);
        if (trustCertCollectionFile != null) {
            sslContext.trustManager(trustCertCollectionFile);
        }
        channel = builder.sslContext(sslContext.build())
                .negotiationType(NegotiationType.TLS)
                .build();
    } else {
        channel = builder.negotiationType(NegotiationType.PLAINTEXT)
                .build();
    }
    retryExecutor = Executors.newSingleThreadScheduledExecutor(
            ThreadFactories.create("Glowroot-Collector-Retry"));
    this.inConnectionFailure = inConnectionFailure;
    this.collectorAddress = collectorAddress;
}

@Override
protected ManagedChannel createChannel() {
  AbstractManagedChannelImplBuilder<?> builder;
  if (!useOkHttp) {
    SslContext sslContext = null;
    if (useTestCa) {
      try {
        sslContext = GrpcSslContexts.forClient().trustManager(
                TestUtils.loadCert("ca.pem")).build();
      } catch (Exception ex) {
        throw new RuntimeException(ex);
      }
    }
    NettyChannelBuilder nettyBuilder =
        NettyChannelBuilder.forAddress(serverHost, serverPort)
            .flowControlWindow(65 * 1024)
            .negotiationType(useTls ? NegotiationType.TLS : NegotiationType.PLAINTEXT)
            .sslContext(sslContext);
    if (serverHostOverride != null) {
      nettyBuilder.overrideAuthority(serverHostOverride);
    }
    if (fullStreamDecompression) {
      nettyBuilder.enableFullStreamDecompression();
    }
    builder = nettyBuilder;
  } else {
    OkHttpChannelBuilder okBuilder = OkHttpChannelBuilder.forAddress(serverHost, serverPort);
    if (serverHostOverride != null) {
      // Force the hostname to match the cert the server uses.
      okBuilder.overrideAuthority(
          GrpcUtil.authorityFromHostAndPort(serverHostOverride, serverPort));
    }
    if (useTls) {
      try {
        SSLSocketFactory factory = useTestCa
            ? TestUtils.newSslSocketFactoryForCa(Platform.get().getProvider(),
                TestUtils.loadCert("ca.pem"))
            : (SSLSocketFactory) SSLSocketFactory.getDefault();
        okBuilder.sslSocketFactory(factory);
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    } else {
      okBuilder.usePlaintext(true);
    }
    if (fullStreamDecompression) {
      okBuilder.enableFullStreamDecompression();
    }
    builder = okBuilder;
  }
  io.grpc.internal.TestingAccessor.setStatsImplementation(
      builder, createClientCensusStatsModule());
  return builder.build();
}

private static NettyChannelBuilder newNettyClientChannel(Transport transport,
    SocketAddress address, boolean tls, boolean testca, int flowControlWindow,
    boolean useDefaultCiphers) throws IOException {
  NettyChannelBuilder builder =
      NettyChannelBuilder.forAddress(address).flowControlWindow(flowControlWindow);
  if (tls) {
    builder.negotiationType(NegotiationType.TLS);
    SslContext sslContext = null;
    if (testca) {
      File cert = TestUtils.loadCert("ca.pem");
      SslContextBuilder sslContextBuilder = GrpcSslContexts.forClient().trustManager(cert);
      if (transport == Transport.NETTY_NIO) {
        sslContextBuilder = GrpcSslContexts.configure(sslContextBuilder, SslProvider.JDK);
      } else {
        // Native transport with OpenSSL
        sslContextBuilder = GrpcSslContexts.configure(sslContextBuilder, SslProvider.OPENSSL);
      }
      if (useDefaultCiphers) {
        sslContextBuilder.ciphers(null);
      }
      sslContext = sslContextBuilder.build();
    }
    builder.sslContext(sslContext);
  } else {
    builder.negotiationType(NegotiationType.PLAINTEXT);
  }

  DefaultThreadFactory tf = new DefaultThreadFactory("client-elg-", true /*daemon */);
  switch (transport) {
    case NETTY_NIO:
      builder
          .eventLoopGroup(new NioEventLoopGroup(0, tf))
          .channelType(NioSocketChannel.class);
      break;

    case NETTY_EPOLL:
      // These classes only work on Linux.
      builder
          .eventLoopGroup(new EpollEventLoopGroup(0, tf))
          .channelType(EpollSocketChannel.class);
      break;

    case NETTY_UNIX_DOMAIN_SOCKET:
      // These classes only work on Linux.
      builder
          .eventLoopGroup(new EpollEventLoopGroup(0, tf))
          .channelType(EpollDomainSocketChannel.class);
      break;

    default:
      // Should never get here.
      throw new IllegalArgumentException("Unsupported transport: " + transport);
  }
  return builder;
}