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

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 GRPCAgentClient build() {

            ManagedChannel managedChannel = channel;

            if (managedChannel == null) {
                managedChannel = NettyChannelBuilder.forAddress(host, port)
                    .keepAliveTime(keepAliveTimeMS, TimeUnit.MILLISECONDS)
                    .keepAliveTimeout(keepAliveTimeoutMS, TimeUnit.MILLISECONDS)
                    .keepAliveWithoutCalls(keepAliveWithoutCalls)
                    .negotiationType(negotiationType)
                    .build();
            }

            SpanAgentStub stub = SpanAgentGrpc.newStub(managedChannel);

            return new GRPCAgentClient(format, managedChannel, stub, observer, shutdownTimeoutMS);
        }

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

            }
        });
}

/**
 * Before the test has started, create the server and channel.
 */
@Override
protected void before() throws Throwable {
    serviceRegistry = new MutableHandlerRegistry();

    NettyServerBuilder serverBuilder = NettyServerBuilder
            .forPort(0)
            .fallbackHandlerRegistry(serviceRegistry);

    if (useDirectExecutor) {
        serverBuilder.directExecutor();
    }

    configureServerBuilder.accept(serverBuilder);
    server = serverBuilder.build().start();
    port = server.getPort();

    NettyChannelBuilder channelBuilder = NettyChannelBuilder.forAddress("localhost", port).usePlaintext(true);
    configureChannelBuilder.accept(channelBuilder);
    channel = channelBuilder.build();
}

private ManagedChannelBuilder<?> defaultChannelBuilder() {
  NettyChannelBuilder channelBuilder = NettyChannelBuilder.forTarget("etcd");

  if (builder.sslContext() != null) {
    channelBuilder.sslContext(builder.sslContext());
  } else {
    channelBuilder.usePlaintext(true);
  }

  channelBuilder.nameResolverFactory(
      forEndpoints(
        Optional.ofNullable(builder.authority()).orElse("etcd"),
        builder.endpoints(),
        Optional.ofNullable(builder.uriResolverLoader())
            .orElseGet(URIResolverLoader::defaultLoader)
      )
  );

  if (builder.loadBalancerFactory() != null) {
    channelBuilder.loadBalancerFactory(builder.loadBalancerFactory());
  }

  channelBuilder.intercept(new AuthTokenInterceptor());

  return channelBuilder;
}

public RemoteDataSet(final HostAndPort serverEndpoint, final int remoteHandle) {
    this.serverEndpoint = serverEndpoint;
    this.remoteHandle = remoteHandle;
    final ExecutorService executorService =
            ExecutorUtils.newNamedThreadPool("remote-data-set:" + serverEndpoint, 5);
    // Using PollSelectorProvider() to avoid Epoll CPU utilization problems.
    // See: https://github.com/netty/netty/issues/327
    final EventLoopGroup workerElg = new NioEventLoopGroup(1,
            ExecutorUtils.newFastLocalThreadFactory("worker"), new PollSelectorProvider());
    this.stub = HillviewServerGrpc.newStub(NettyChannelBuilder
            .forAddress(serverEndpoint.getHost(), serverEndpoint.getPort())
            .maxInboundMessageSize(HillviewServer.MAX_MESSAGE_SIZE)
            .executor(executorService)
            .eventLoopGroup(workerElg)
            .usePlaintext(true)   // channel is unencrypted.
            .build());
}

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

@Test
public void mockGrpcServerServesOverSSL() {
  final NettyChannelBuilder channelBuilder =
      NettyChannelBuilder.forTarget(storageProperties.getApiHost());

  final TraceServiceBlockingStub plainTraceService =
      TraceServiceGrpc.newBlockingStub(channelBuilder.build());
  final TraceServiceBlockingStub sslTraceService =
      TraceServiceGrpc.newBlockingStub(channelBuilder.sslContext(CLIENT_SSL_CONTEXT).build());

  try {
    plainTraceService.patchTraces(PatchTracesRequest.getDefaultInstance());
  } catch (StatusRuntimeException e) {
    assertThat(e.getMessage(), endsWith("Channel closed while performing protocol negotiation"));
  }

  sslTraceService.patchTraces(PatchTracesRequest.getDefaultInstance());
}

@Before
public void setUp() throws Throwable {
    List<String> grpcServers = new ArrayList<String>();
    grpcServers.add("127.0.0.1:2181");
    RemoteDownstreamConfig.Collector.GRPC_SERVERS = grpcServers;
    Config.Collector.GRPC_CHANNEL_CHECK_INTERVAL = 1;

    mockStatic(NettyChannelBuilder.class);
    when(NettyChannelBuilder.forAddress(anyString(), anyInt())).thenReturn(mock);
    when(mock.nameResolverFactory(any(NameResolver.Factory.class))).thenReturn(mock);
    when(mock.maxInboundMessageSize(anyInt())).thenReturn(mock);
    when(mock.usePlaintext(true)).thenReturn(mock);
    when(mock.build()).thenReturn(grpcServerRule.getChannel());

    grpcChannelManager.addChannelListener(listener);
}

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

/**
 * Create a P4Runtime client for this device. Returns true if the operation was successful, false otherwise.
 *
 * @return true if successful, false otherwise
 */
protected boolean createClient() {
    deviceId = handler().data().deviceId();
    controller = handler().get(P4RuntimeController.class);

    String serverAddr = this.data().value(P4RUNTIME_SERVER_ADDR_KEY);
    String serverPortString = this.data().value(P4RUNTIME_SERVER_PORT_KEY);
    String p4DeviceIdString = this.data().value(P4RUNTIME_DEVICE_ID_KEY);

    if (serverAddr == null || serverPortString == null || p4DeviceIdString == null) {
        log.warn("Unable to create client for {}, missing driver data key (required is {}, {}, and {})",
                 deviceId, P4RUNTIME_SERVER_ADDR_KEY, P4RUNTIME_SERVER_PORT_KEY, P4RUNTIME_DEVICE_ID_KEY);
        return false;
    }

    ManagedChannelBuilder channelBuilder = NettyChannelBuilder
            .forAddress(serverAddr, Integer.valueOf(serverPortString))
            .usePlaintext(true);

    if (!controller.createClient(deviceId, Long.parseUnsignedLong(p4DeviceIdString), channelBuilder)) {
        log.warn("Unable to create client for {}, aborting operation", deviceId);
        return false;
    }

    return true;
}

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

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

public PeerConnection(URI nodeUrl) {
    Protocol protocol = Protocol.valueOf(nodeUrl.getScheme());
    channel = NettyChannelBuilder
        .forAddress(nodeUrl.getHost(), nodeUrl.getPort())
        .usePlaintext(protocol == Protocol.grpc)
        .build();
    stateChannel = StateChannelGrpc.newBlockingStub(channel);
}

private HelloWorldClient(String host, int port, TlsContext tls) throws SSLException {
  this.eventLoopGroup = Netty.newWorkerEventLoopGroup();
  this.channel =
      NettyChannelBuilder.forAddress(host, port)
          .eventLoopGroup(eventLoopGroup)
          .channelType(Netty.clientChannelType())
          .sslContext(tls.toClientContext())
          .build();
  this.blockingStub = GreeterGrpc.newBlockingStub(channel);
}

@Test
public void nettyChannel(){
    ManagedChannel channel = NettyChannelBuilder.forAddress("127.0.0.1",50051).usePlaintext(true).build();
    GreeterGrpc.GreeterBlockingStub blockingStub = GreeterGrpc.newBlockingStub(channel);
    HelloRequest request = HelloRequest.newBuilder().setName("gggg").build();
    HelloReply response;
    blockingStub.sayHello(request);

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

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

public static void main(String... args) throws Exception {
  System.out.println("starting server");

  String host = System.getProperty("host", "0.0.0.0");
  int port = Integer.getInteger("port", 8001);
  boolean useEpoll = Boolean.getBoolean("usePoll");

  Class channel;

  if (useEpoll) {
    channel = EpollServerSocketChannel.class;
  } else  {
    channel = NioServerSocketChannel.class;
  }

  ThreadFactory tf = new DefaultThreadFactory("server-elg-", true /*daemon */);
  NioEventLoopGroup boss = new NioEventLoopGroup(1, tf);
  NioEventLoopGroup worker = new NioEventLoopGroup(0, tf);
  NettyServerBuilder builder =
      NettyServerBuilder.forPort(port)
          .bossEventLoopGroup(boss)
          .workerEventLoopGroup(worker)
          .channelType(channel)
          .addService(new DefaultService())
          .directExecutor()
          .maxConcurrentCallsPerConnection(Runtime.getRuntime().availableProcessors() * 256)
          .flowControlWindow(NettyChannelBuilder.DEFAULT_FLOW_CONTROL_WINDOW * 10);

  io.grpc.Server start = builder.build();
  start.start();

  System.out.println("server started");
  start.awaitTermination();
}

public Channel createChannel(HostAndPort endpoint) {
  NettyChannelBuilder nettyChannelBuilder = createChannelBuilder(endpoint);

  if (!callConfiguration.getTlsClientOverrideAuthority().isEmpty()) {
    nettyChannelBuilder.overrideAuthority(callConfiguration.getTlsClientOverrideAuthority());
  }

  return nettyChannelBuilder.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 TensorflowObjectRecogniser(String host, int port) {
  LOG.debug("Creating channel host:{}, port={}", host, port);
  try {
    channel = NettyChannelBuilder
        .forAddress(host, port)
        .usePlaintext(true)
        .build();
    stub = new InceptionBlockingStub(channel);
    //TODO: test channel here with a sample image
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
}

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

public ADAMproWrapper() {
  DatabaseConfig config = Config.sharedConfig().getDatabase();
  this.channel = NettyChannelBuilder.forAddress(config.getHost(), config.getPort())
      .maxMessageSize(maxMessageSize).usePlaintext(config.getPlaintext()).build();
  this.definitionStub = AdamDefinitionGrpc.newFutureStub(channel);
  this.searchStub = AdamSearchGrpc.newFutureStub(channel);
}

private HelloServiceStub createClientStub(Configuration configuration) {
  return HelloServiceGrpc.newStub(NettyChannelBuilder.forAddress("localhost", grpcPort)
      .usePlaintext(true)
      .intercept(MonitoringClientInterceptor.create(
          configuration.withCollectorRegistry(collectorRegistry)))
      .build());
}

@Test
public void testKeepAlive() throws IOException, ExecutionException, InterruptedException {

    // Verify that keep-alive timeout less than permissible by the server results in a failure.
    final ControllerImpl controller = new ControllerImpl(NettyChannelBuilder.forAddress("localhost", serverPort)
            .keepAliveTime(10, TimeUnit.SECONDS).usePlaintext(true),
            ControllerImplConfig.builder().retryAttempts(1).build(), this.executor);
    CompletableFuture<Boolean> createStreamStatus = controller.createStream(StreamConfiguration.builder()
            .streamName("streamdelayed")
            .scope("scope1")
            .scalingPolicy(ScalingPolicy.fixed(1))
            .build());
    AssertExtensions.assertThrows("Should throw RetriesExhaustedException", createStreamStatus,
            throwable -> throwable instanceof RetriesExhaustedException);

    // Verify that the same RPC with permissible keepalive time succeeds.
    int serverPort2 = TestUtils.getAvailableListenPort();
    Server testServer = NettyServerBuilder.forPort(serverPort2)
            .addService(testServerImpl)
            .permitKeepAliveTime(5, TimeUnit.SECONDS)
            .build()
            .start();
    final ControllerImpl controller1 = new ControllerImpl(NettyChannelBuilder.forAddress("localhost", serverPort2)
            .keepAliveTime(10, TimeUnit.SECONDS).usePlaintext(true),
            ControllerImplConfig.builder().retryAttempts(1).build(), this.executor);
    createStreamStatus = controller1.createStream(StreamConfiguration.builder()
            .streamName("streamdelayed")
            .scope("scope1")
            .scalingPolicy(ScalingPolicy.fixed(1))
            .build());
    assertTrue(createStreamStatus.get());
    testServer.shutdownNow();
}

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

@Override
public ManagedChannel forDescriptor(ApiServiceDescriptor apiServiceDescriptor) {
  SocketAddress address = SocketAddressFactory.createFrom(apiServiceDescriptor.getUrl());
  return NettyChannelBuilder.forAddress(address)
      .channelType(address instanceof DomainSocketAddress
          ? EpollDomainSocketChannel.class : EpollSocketChannel.class)
      .eventLoopGroup(new EpollEventLoopGroup())
      .usePlaintext(true)
      // Set the message size to max value here. The actual size is governed by the
      // buffer size in the layers above.
      .maxInboundMessageSize(Integer.MAX_VALUE)
      .build();
}

@Override
protected ManagedChannel newChannel() {
  return NettyChannelBuilder
      .forAddress("localhost", 1234)
      .flowControlWindow(flowWindow)
      .usePlaintext(true)
      .build();
}

private static EchoGrpcRx.EchoStub newClient() {
  channel = NettyChannelBuilder
      .forAddress("abc", 123)
      .usePlaintext(true)
      .build();

  return EchoGrpcRx.newStub(channel);
}

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

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

@Override
public void run() {
    logger.debug("Selected collector grpc service running, reconnect:{}.", reconnect);
    if (reconnect) {
        if (RemoteDownstreamConfig.Collector.GRPC_SERVERS.size() > 0) {
            String server = "";
            try {
                int index = Math.abs(random.nextInt()) % RemoteDownstreamConfig.Collector.GRPC_SERVERS.size();
                server = RemoteDownstreamConfig.Collector.GRPC_SERVERS.get(index);
                String[] ipAndPort = server.split(":");
                ManagedChannelBuilder<?> channelBuilder =
                    NettyChannelBuilder.forAddress(ipAndPort[0], Integer.parseInt(ipAndPort[1]))
                        .nameResolverFactory(new DnsNameResolverProvider())
                        .maxInboundMessageSize(1024 * 1024 * 50)
                        .usePlaintext(true);
                managedChannel = channelBuilder.build();
                if (!managedChannel.isShutdown() && !managedChannel.isTerminated()) {
                    reconnect = false;
                    notify(GRPCChannelStatus.CONNECTED);
                } else {
                    notify(GRPCChannelStatus.DISCONNECT);
                }
                return;
            } catch (Throwable t) {
                logger.error(t, "Create channel to {} fail.", server);
                notify(GRPCChannelStatus.DISCONNECT);
            }
        }

        logger.debug("Selected collector grpc service is not available. Wait {} seconds to retry", Config.Collector.GRPC_CHANNEL_CHECK_INTERVAL);
    }
}

public static void main(String[] args) throws InterruptedException {
    ManagedChannelBuilder<?> channelBuilder =
        NettyChannelBuilder.forAddress("127.0.0.1", 8080)
            .nameResolverFactory(new DnsNameResolverProvider())
            .maxInboundMessageSize(1024 * 1024 * 50)
            .usePlaintext(true);
    ManagedChannel channel = channelBuilder.build();
    TraceSegmentServiceGrpc.TraceSegmentServiceStub serviceStub = TraceSegmentServiceGrpc.newStub(channel);
    final Status[] status = {null};
    StreamObserver<UpstreamSegment> streamObserver = serviceStub.collect(new StreamObserver<Downstream>() {
        @Override public void onNext(Downstream value) {

        }

        @Override public void onError(Throwable t) {
            status[0] = ((StatusRuntimeException)t).getStatus();
        }

        @Override public void onCompleted() {

        }
    });

    streamObserver.onNext(null);
    streamObserver.onCompleted();

    Thread.sleep(2 * 1000);

    Assert.assertEquals(status[0].getCode(), Status.UNAVAILABLE.getCode());
}

private static ManagedChannel build(String host, int port) {
  return NettyChannelBuilder.forAddress(host, port)
      // this overrides dns lookup, maybe
      // .overrideAuthority(TestUtils.TEST_SERVER_HOST)
      .overrideAuthority(host + ":" + port)
      // this is the default
      // .negotiationType(NegotiationType.TLS)
      .sslContext(
          SslContextFactory.buildClientContext(
              TlsConfig.fromConfig("xio.h2TestClient.settings.tls"),
              InsecureTrustManagerFactory.INSTANCE))
      .build();
}

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