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

@BeforeClass
public static void startServer() throws IOException {
    AfricasTalking.initialize(Fixtures.USERNAME, Fixtures.API_KEY);
    server = new Server(new Authenticator() {
        @Override
        public boolean authenticate(String client) {
            return client.compareToIgnoreCase(TEST_CLIENT_ID) == 0;
        }
    });
    server.addSipCredentials("test", "secret", "sip://at.dev");
    server.start(certFile, privateKeyFile, TEST_PORT);

    ManagedChannel ch = NettyChannelBuilder.forAddress("localhost", TEST_PORT)
        .sslContext(GrpcSslContexts.forClient().trustManager(certFile).build())
        .build();


    client = SdkServerServiceGrpc.newBlockingStub(ch)
        .withCallCredentials(new CallCredentials(){
            @Override
            public void applyRequestMetadata(MethodDescriptor<?, ?> method, Attributes attrs, Executor appExecutor,
                    final MetadataApplier applier) {
                        appExecutor.execute(new Runnable(){
                            @Override
                            public void run() {
                                try {
                                    Metadata headers = new Metadata();
                                    Metadata.Key<String> clientIdKey = Metadata.Key.of("X-Client-Id", Metadata.ASCII_STRING_MARSHALLER);
                                    headers.put(clientIdKey, TEST_CLIENT_ID);
                                    applier.apply(headers);
                                } catch(Throwable ex) {
                                    applier.fail(Status.UNAUTHENTICATED.withCause(ex));
                                }
                            }
                        });

            }
        });
}

private synchronized static SslContext createSslContext() throws SSLException {
  if (sslBuilder == null) {
    sslBuilder = GrpcSslContexts.forClient().ciphers(null);
    // gRPC uses tcnative / OpenSsl by default, if it's available. It defaults to alpn-boot
    // if tcnative is not in the classpath.
    if (OpenSsl.isAvailable()) {
      LOG.info(
          "SslContext: gRPC is using the OpenSSL provider (tcnactive jar - Open Ssl version: {})",
          OpenSsl.versionString());
    } else {
      if (isJettyAlpnConfigured()) {
        // gRPC uses jetty ALPN as a backup to tcnative.
        LOG.info("SslContext: gRPC is using the JDK provider (alpn-boot jar)");
      } else {
        LOG.info("SslContext: gRPC cannot be configured.  Neither OpenSsl nor Alpn are available.");
      }
    }
  }
  return sslBuilder.build();
}

@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());
}

/** Starts the server with HTTPS. */
@BeforeClass
public static void startServer() throws Exception {
    SLF4JBridgeHandler.removeHandlersForRootLogger();
    SLF4JBridgeHandler.install();

    ssc = new SelfSignedCertificate("example.com");
    ServerBuilder sb = new ServerBuilder()
            .port(0, SessionProtocol.HTTPS)
            .defaultMaxRequestLength(16 * 1024 * 1024)
            .sslContext(GrpcSslContexts.forServer(ssc.certificate(), ssc.privateKey())
                                       .applicationProtocolConfig(ALPN)
                                       .trustManager(TestUtils.loadCert("ca.pem"))
                                       .build());

    final ArmeriaGrpcServerBuilder builder = new ArmeriaGrpcServerBuilder(sb, new GrpcServiceBuilder(),
                                                                          ctxCapture);
    startStaticServer(builder);
    server = builder.builtServer();
}

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();
}

@VisibleForTesting
void start() throws Exception {
  executor = Executors.newSingleThreadScheduledExecutor();
  SslContext sslContext = null;
  if (useTls) {
    sslContext = GrpcSslContexts.forServer(
            TestUtils.loadCert("server1.pem"), TestUtils.loadCert("server1.key")).build();
  }
  server = NettyServerBuilder.forPort(port)
      .sslContext(sslContext)
      .maxMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
      .addService(ServerInterceptors.intercept(
          new TestServiceImpl(executor),
          TestServiceImpl.interceptors()))
      .build().start();
}

/**
 * Creates and starts a new {@link TestServiceImpl} server.
 */
private Server newServer() throws CertificateException, IOException {
  File serverCertChainFile = TestUtils.loadCert("server1.pem");
  File serverPrivateKeyFile = TestUtils.loadCert("server1.key");
  X509Certificate[] serverTrustedCaCerts = {
    TestUtils.loadX509Cert("ca.pem")
  };

  SslContext sslContext =
      GrpcSslContexts.forServer(serverCertChainFile, serverPrivateKeyFile)
                     .trustManager(serverTrustedCaCerts)
                     .clientAuth(ClientAuth.REQUIRE)
                     .build();

  return NettyServerBuilder.forPort(0)
      .sslContext(sslContext)
      .addService(new TestServiceImpl(serverExecutor))
      .build()
      .start();
}

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();
}

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  // Starts the server with HTTPS.
  try {
    SslProvider sslProvider = SslContext.defaultServerProvider();
    if (sslProvider == SslProvider.OPENSSL && !OpenSsl.isAlpnSupported()) {
      // OkHttp only supports Jetty ALPN on OpenJDK. So if OpenSSL doesn't support ALPN, then we
      // are forced to use Jetty ALPN for Netty instead of OpenSSL.
      sslProvider = SslProvider.JDK;
    }
    SslContextBuilder contextBuilder = SslContextBuilder
        .forServer(TestUtils.loadCert("server1.pem"), TestUtils.loadCert("server1.key"));
    GrpcSslContexts.configure(contextBuilder, sslProvider);
    contextBuilder.ciphers(TestUtils.preferredTestCiphers(), SupportedCipherSuiteFilter.INSTANCE);
    return NettyServerBuilder.forPort(0)
        .flowControlWindow(65 * 1024)
        .maxMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
        .sslContext(contextBuilder.build());
  } catch (IOException ex) {
    throw new RuntimeException(ex);
  }
}

@Before
public void setUp() throws NoSuchAlgorithmException {
  executor = Executors.newSingleThreadScheduledExecutor();
  if (sslProvider == SslProvider.OPENSSL) {
    Assume.assumeTrue(OpenSsl.isAvailable());
  }
  if (sslProvider == SslProvider.JDK) {
    Assume.assumeTrue(Arrays.asList(
        SSLContext.getDefault().getSupportedSSLParameters().getCipherSuites())
        .contains("TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256"));
    try {
      GrpcSslContexts.configure(SslContextBuilder.forClient(), SslProvider.JDK);
    } catch (IllegalArgumentException ex) {
      Assume.assumeNoException("Jetty ALPN does not seem available", ex);
    }
  }
  clientContextBuilder = GrpcSslContexts.configure(SslContextBuilder.forClient(), sslProvider);
}

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  // Starts the server with HTTPS.
  try {
    return NettyServerBuilder.forPort(0)
        .flowControlWindow(65 * 1024)
        .maxMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
        .sslContext(GrpcSslContexts
            .forServer(TestUtils.loadCert("server1.pem"), TestUtils.loadCert("server1.key"))
            .clientAuth(ClientAuth.REQUIRE)
            .trustManager(TestUtils.loadCert("ca.pem"))
            .ciphers(TestUtils.preferredTestCiphers(), SupportedCipherSuiteFilter.INSTANCE)
            .sslProvider(SslProvider.OPENSSL)
            .build());
  } catch (IOException ex) {
    throw new RuntimeException(ex);
  }
}

@Override
protected ManagedChannel createChannel() {
  try {
    NettyChannelBuilder builder = NettyChannelBuilder
        .forAddress(TestUtils.testServerAddress(getPort()))
        .flowControlWindow(65 * 1024)
        .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
        .sslContext(GrpcSslContexts
            .forClient()
            .keyManager(TestUtils.loadCert("client.pem"), TestUtils.loadCert("client.key"))
            .trustManager(TestUtils.loadX509Cert("ca.pem"))
            .ciphers(TestUtils.preferredTestCiphers(), SupportedCipherSuiteFilter.INSTANCE)
            .sslProvider(SslProvider.OPENSSL)
            .build());
    io.grpc.internal.TestingAccessor.setStatsImplementation(
        builder, createClientCensusStatsModule());
    return builder.build();
  } catch (Exception ex) {
    throw new RuntimeException(ex);
  }
}

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;
}

private void configureSsl(NettyServerBuilder builder) throws NoSuchAlgorithmException, CertificateEncodingException, NoSuchProviderException, InvalidKeyException, SignatureException, SSLException {
    NativeLibraryLoader.loadFirstAvailable(ClassLoader.getSystemClassLoader(),
        "netty_tcnative_osx_x86_64",
        "netty_tcnative_linux_x86_64",
        "netty_tcnative_windows_x86_64"
    );
    ECKeyPair ecKeyPair = ethereumConfig.getMainCredentials().getEcKeyPair();
    KeyPair keyPair = CryptoUtil.decodeKeyPair(ecKeyPair);
    SslContextBuilder contextBuilder = SslContextBuilder.forServer(
        keyPair.getPrivate(),
        CryptoUtil.genCert(keyPair)
    );

    builder.sslContext(GrpcSslContexts.configure(contextBuilder).build());
}

private SslContext buildClientSslContext() {
  try {
    InputStream certs = SslUtil.loadInputStreamCert("server.pem");
    return GrpcSslContexts
        .configure(SslContextBuilder.forClient()//
            .trustManager(certs))//
        .build();
  } catch (SSLException e) {
    throw new RpcFrameworkException(e);
  }
}

private SslContext buildServerSslContext() {
  try {
    InputStream certs = SslUtil.loadInputStreamCert("server.pem");
    InputStream keys = SslUtil.loadInputStreamCert("server_pkcs8.key");
    return GrpcSslContexts.configure(SslContextBuilder.forServer(certs, keys)).build();
  } catch (SSLException e) {
    throw new RpcFrameworkException(e);
  }
}

@Test(timeout = 5000)
public void testSimpleSllSetup() throws Exception {
  final ByteSequence key = ByteSequence.fromString(TestUtil.randomString());
  final ByteSequence val = ByteSequence.fromString(TestUtil.randomString());
  final String capath = System.getProperty("ssl.cert.capath");
  final String authority = System.getProperty("ssl.cert.authority", TestConstants.DEFAULT_SSL_AUTHORITY);
  final String endpoints = System.getProperty("ssl.cert.endpoints", TestConstants.DEFAULT_SSL_ENDPOINTS);

  try (InputStream is = Objects.nonNull(capath)
        ? new FileInputStream(new File(capath))
        : getClass().getResourceAsStream(TestConstants.DEFAULT_SSL_CA_PATH)) {

    Client client = Client.builder()
        .endpoints(endpoints)
        .authority(authority)
        .sslContext(GrpcSslContexts.forClient()
            .trustManager(is)
            .build())
        .build();

    KV kv = client.getKVClient();
    kv.put(key, val).join();

    assertThat(kv.get(key).join().getCount()).isEqualTo(1);
    assertThat(kv.get(key).join().getKvs().get(0).getValue()).isEqualTo(val);

    kv.close();
    client.close();
  }
}

/**
 * Build NettyServerBuilder.
 * 1. Enable tls or not
 * 2. Load gRPC specified service definition and then add to gRPC registry.
 */
public Server buildServer() throws IOException {

  SslContext sslContext;
  NettyServerBuilder serverBuilder = NettyServerBuilder.forPort(port).executor(Executors.newFixedThreadPool(8));
  if (useTls) {
    sslContext = GrpcSslContexts
        .forServer(SslUtils.loadCert("server.pem"), SslUtils.loadCert("server.key")).build();
    serverBuilder.sslContext(sslContext);
  }
  serverBuilder = bindService(serverBuilder);

  return serverBuilder.flowControlWindow(1024 * 1024).build();// 1024*1024 = 1MiB
}

/**
 * Creates a <code>ManagedChannel</code> connecting to an <b>encrypted</b> gRPC server in
 * <code>TestApplication</code> in <code>testSupport</code>. The certificate is taken from the
 * <code>GrpcServerFactory</code> in the configuration.
 *
 * @param testSupport the already initialised (started) <code>DropwizardTestSupport</code> instance
 * @return the channel connecting to the server (to be used in a client)
 */
public static ManagedChannel createClientChannelForEncryptedServer(
        final DropwizardTestSupport<TestConfiguration> testSupport) throws SSLException {
    final SslContext sslContext = GrpcSslContexts.forClient()
        .trustManager(testSupport.getConfiguration().getGrpcServerFactory().getCertChainFile().toFile()).build();
    final TestApplication application = testSupport.getApplication();
    return NettyChannelBuilder.forAddress("localhost", application.getServer().getPort()).sslContext(sslContext)
        .overrideAuthority("grpc-dropwizard.example.com").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();
}

private ServerBuilder<?> serverBuilder(int port, File serverCertChainFile,
    File serverPrivateKeyFile, X509Certificate[] serverTrustedCaCerts) throws IOException {
  SslContextBuilder sslContextBuilder
      = SslContextBuilder.forServer(serverCertChainFile, serverPrivateKeyFile);
  GrpcSslContexts.configure(sslContextBuilder, sslProvider);
  sslContextBuilder.trustManager(serverTrustedCaCerts)
      .clientAuth(ClientAuth.REQUIRE);

  return NettyServerBuilder.forPort(port)
      .sslContext(sslContextBuilder.build());
}

public SslContext toClientContext() throws SSLException {
  return GrpcSslContexts.configure(SslContextBuilder.forClient(), SslProvider.OPENSSL)
      .trustManager(trustedCerts)
      .keyManager(cert, key)
      .build();
}

public SslContext toServerContext() throws SSLException {
  return GrpcSslContexts.configure(SslContextBuilder.forServer(cert, key), SslProvider.OPENSSL)
      .trustManager(trustedCerts)
      .clientAuth(ClientAuth.REQUIRE)
      .build();
}

private static SslContextBuilder getSslContextBuilder() {
  return GrpcSslContexts.forServer(TestUtils.loadServerChainCert(), TestUtils.loadServerKey())
      .trustManager(TestUtils.loadRootCaCert())
      .sslProvider(SslProvider.OPENSSL);
}

/**
 * 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();
}

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;
}