import io.grpc.ClientInterceptor; //导入依赖的package包/类
private ClientInterceptor metadataInterceptor() {
  ClientInterceptor interceptor = new ClientInterceptor() {
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        final io.grpc.MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, final Channel next) {
      return new ClientInterceptors.CheckedForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
        @Override
        protected void checkedStart(Listener<RespT> responseListener, Metadata headers)
            throws StatusException {
          for (ConfigProto.CallMetadataEntry entry : callConfiguration.getMetadataList()) {
            Metadata.Key<String> key = Metadata.Key.of(entry.getName(), Metadata.ASCII_STRING_MARSHALLER);
            headers.put(key, entry.getValue());
          }
          delegate().start(responseListener, headers);
        }
      };
    }
  };

  return interceptor;
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
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();
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
/** Construct client for accessing RouteGuide server using the existing channel. */
public SeldonClientExample(ManagedChannelBuilder<?> channelBuilder) {
 ClientInterceptor interceptor = new HeaderClientInterceptor();
  channel = channelBuilder.build();
  Channel interceptChannel = ClientInterceptors.intercept(channel, interceptor);
  blockingStub = SeldonGrpc.newBlockingStub(interceptChannel);
  asyncStub = SeldonGrpc.newStub(interceptChannel);
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
@VisibleForTesting
final List<ClientInterceptor> getEffectiveInterceptors() {
  List<ClientInterceptor> effectiveInterceptors =
      new ArrayList<ClientInterceptor>(this.interceptors);
  if (statsEnabled) {
    CensusStatsModule censusStats = this.censusStatsOverride;
    if (censusStats == null) {
      censusStats = new CensusStatsModule(GrpcUtil.STOPWATCH_SUPPLIER, true);
    }
    // First interceptor runs last (see ClientInterceptors.intercept()), so that no
    // other interceptor can override the tracer factory we set in CallOptions.
    effectiveInterceptors.add(
        0, censusStats.getClientInterceptor(recordStartedRpcs, recordFinishedRpcs));
  }
  if (tracingEnabled) {
    CensusTracingModule censusTracing =
        new CensusTracingModule(Tracing.getTracer(),
            Tracing.getPropagationComponent().getBinaryFormat());
    effectiveInterceptors.add(0, censusTracing.getClientInterceptor());
  }
  return effectiveInterceptors;
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
    MethodDescriptor<ReqT, RespT> method,
    CallOptions callOptions,
    Channel next) {
  ClientInterceptor binlogInterceptor = getClientInterceptor(method.getFullMethodName());
  if (binlogInterceptor == null) {
    return next.newCall(method, callOptions);
  } else {
    return InternalClientInterceptors
        .wrapClientInterceptor(
            binlogInterceptor,
            IDENTITY_MARSHALLER,
            IDENTITY_MARSHALLER)
        .interceptCall(method, callOptions, next);
  }
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
/**
 * A custom client.
 */
private CustomHeaderClient(String host, int port) {
  originChannel = ManagedChannelBuilder.forAddress(host, port)
      .usePlaintext(true)
      .build();
  ClientInterceptor interceptor = new HeaderClientInterceptor();
  Channel channel = ClientInterceptors.intercept(originChannel, interceptor);
  blockingStub = GreeterGrpc.newBlockingStub(channel);
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
public GrpcClientInitializer(ClientOptions pClientOptions,
    List<ClientInterceptor> clientInterceptosr, int pInitialCapacity, int pMaximumSize) {
  LOG.info("Rpc client initializer with initial capacity {} and maximum size {} for channel pool.",
      pInitialCapacity, pInitialCapacity);
  LOG.info("Global client options: \n'{}'.", pClientOptions);
  
  if (!isAlpnProviderEnabled()) {
    LOG.error(
        "Neither Jetty ALPN nor OpenSSL are available. "
        + "OpenSSL unavailability cause:\n{}",
        OpenSsl.unavailabilityCause().toString());
    throw new IllegalStateException("Neither Jetty ALPN nor OpenSSL via "
        + "netty-tcnative were properly configured.");
  }
  
  Preconditions
  .checkState(
      !AbstractNameResolverProvider.providers().isEmpty(),
      "No NameResolverProviders found via ServiceLoader, including for DNS. "
          + "This is probably due to a broken build. If using ProGuard, check your configuration");
  
  globalClientOptions = pClientOptions;
  
  channelPool = createChannelPool(globalClientOptions, clientInterceptosr, pInitialCapacity, pMaximumSize);
  
  ClientMetrics.counter(MetricLevel.Info, "Initializer.active").inc();
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
/**
 * Create a new {@link com.github.ibole.microservice.rpc.client.grpc.ChannelPool}.
 *
 * @param pInitialCapacity
 * @param pMaximumSize
 * @param globalClientOptions a {@link ClientOptions} object with registry center server address and other connection options.
 * @param interceptors a list of interceptor
 * @return a {@link ChannelPool} object.
 */
private ChannelPool createChannelPool(ClientOptions globalClientOptions, List<ClientInterceptor> interceptors, int pInitialCapacity, int pMaximumSize) {
  return ChannelPool.newBuilder().withChannelFactory(new ChannelPool.ChannelFactory() {
    @Override
    public ManagedChannel create(String serviceName, String preferredZone, boolean usedTls) throws IOException {
      //build service endpoint with the default scheme and the service name provided
      String serviceEndpoint = AbstractNameResolverProvider.provider().getDefaultScheme() + "://" + serviceName;
      return createNettyChannel(globalClientOptions.withServiceEndpoint(serviceEndpoint).withZoneToPrefer(preferredZone).withUsedTls(usedTls), interceptors);
    }
  }).withInitialCapacity(pInitialCapacity).withMaximumSize(pMaximumSize).build();
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
/**
 * Create a new gRPC channel to the Google Genomics API, using the provided credentials for auth.
 *
 * @param creds The credential.
 * @param fields Which fields to return in the partial response, or null for none.
 * @return The ManagedChannel.
 * @throws SSLException
 */
public static ManagedChannel fromCreds(GoogleCredentials creds, String fields) throws SSLException {
  List<ClientInterceptor> interceptors = new ArrayList();
  interceptors.add(new ClientAuthInterceptor(creds.createScoped(Arrays.asList(GENOMICS_SCOPE)),
      Executors.newSingleThreadExecutor()));
  if (!Strings.isNullOrEmpty(fields)) {
    Metadata headers = new Metadata();
    Metadata.Key<String> partialResponseHeader =
        Metadata.Key.of(PARTIAL_RESPONSE_HEADER, Metadata.ASCII_STRING_MARSHALLER);
    headers.put(partialResponseHeader, fields);
    interceptors.add(MetadataUtils.newAttachHeadersInterceptor(headers));
  }
  return getGenomicsManagedChannel(interceptors);
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
@Test
public void serverHeaderDeliveredToClient() {
  class SpyingClientInterceptor implements ClientInterceptor {
    ClientCall.Listener<?> spyListener;

    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
      return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
        @Override
        public void start(Listener<RespT> responseListener, Metadata headers) {
          spyListener = responseListener =
              mock(ClientCall.Listener.class, delegatesTo(responseListener));
          super.start(responseListener, headers);
        }
      };
    }
  }

  SpyingClientInterceptor clientInterceptor = new SpyingClientInterceptor();
  GreeterBlockingStub blockingStub = GreeterGrpc.newBlockingStub(grpcServerRule.getChannel())
      .withInterceptors(clientInterceptor);
  ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);

  blockingStub.sayHello(HelloRequest.getDefaultInstance());

  assertNotNull(clientInterceptor.spyListener);
  verify(clientInterceptor.spyListener).onHeaders(metadataCaptor.capture());
  assertEquals(
      "customRespondValue",
      metadataCaptor.getValue().get(HeaderServerInterceptor.CUSTOM_HEADER_KEY));
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
@Test
public void getEffectiveInterceptors_default() {
  builder.intercept(DUMMY_USER_INTERCEPTOR);
  List<ClientInterceptor> effectiveInterceptors = builder.getEffectiveInterceptors();
  assertEquals(3, effectiveInterceptors.size());
  assertThat(effectiveInterceptors.get(0))
      .isInstanceOf(CensusTracingModule.TracingClientInterceptor.class);
  assertThat(effectiveInterceptors.get(1))
      .isInstanceOf(CensusStatsModule.StatsClientInterceptor.class);
  assertThat(effectiveInterceptors.get(2)).isSameAs(DUMMY_USER_INTERCEPTOR);
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
@Test
public void getEffectiveInterceptors_disableStats() {
  builder.intercept(DUMMY_USER_INTERCEPTOR);
  builder.setStatsEnabled(false);
  List<ClientInterceptor> effectiveInterceptors = builder.getEffectiveInterceptors();
  assertEquals(2, effectiveInterceptors.size());
  assertThat(effectiveInterceptors.get(0))
      .isInstanceOf(CensusTracingModule.TracingClientInterceptor.class);
  assertThat(effectiveInterceptors.get(1)).isSameAs(DUMMY_USER_INTERCEPTOR);
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
@Test
public void getEffectiveInterceptors_disableTracing() {
  builder.intercept(DUMMY_USER_INTERCEPTOR);
  builder.setTracingEnabled(false);
  List<ClientInterceptor> effectiveInterceptors = builder.getEffectiveInterceptors();
  assertEquals(2, effectiveInterceptors.size());
  assertThat(effectiveInterceptors.get(0))
      .isInstanceOf(CensusStatsModule.StatsClientInterceptor.class);
  assertThat(effectiveInterceptors.get(1)).isSameAs(DUMMY_USER_INTERCEPTOR);
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
@Test
public void getEffectiveInterceptors_disableBoth() {
  builder.intercept(DUMMY_USER_INTERCEPTOR);
  builder.setStatsEnabled(false);
  builder.setTracingEnabled(false);
  List<ClientInterceptor> effectiveInterceptors = builder.getEffectiveInterceptors();
  assertThat(effectiveInterceptors).containsExactly(DUMMY_USER_INTERCEPTOR);
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
public static ClientInterceptor instance() {
  return new HeaderClientInterceptor();
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
/**
 * 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);
                }
            });

}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
/**
 * <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();
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
private InterceptorChannel(Channel channel, ClientInterceptor interceptor) {
  this.channel = channel;
  this.interceptor = Preconditions.checkNotNull(interceptor, "interceptor");
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
private static CloseableChannel wrapChannel(ChannelOptions channelOptions,
    ExecutorService executor, Channel channel, ClientCloseHandler onClientClose) {
  List<ClientInterceptor> interceptors = new ArrayList<>();
  if (channelOptions.getCredential() != null) {
    interceptors.add(new ClientAuthInterceptor(channelOptions.getCredential(), executor));
  }

  if (channelOptions.getAuthority() != null) {
    Metadata.Headers headers = new Metadata.Headers();
    headers.setAuthority(channelOptions.getAuthority());
    interceptors.add(MetadataUtils.newAttachHeadersInterceptor(headers));
  }

  CallCompletionStatusInterceptor preRetryCallStatusInterceptor = null;
  if (!Strings.isNullOrEmpty(channelOptions.getCallStatusReportPath())) {
    preRetryCallStatusInterceptor = new CallCompletionStatusInterceptor();
    interceptors.add(preRetryCallStatusInterceptor);
  }

  if (!interceptors.isEmpty()) {
    channel = ClientInterceptors.intercept(channel, interceptors);
    interceptors.clear();
  }

  if (channelOptions.getUnaryCallRetryOptions().enableRetries()) {
    ScheduledExecutorService scheduledRetries;
    if (channelOptions.getScheduledExecutorService() != null) {
      scheduledRetries = channelOptions.getScheduledExecutorService();
    } else {
      scheduledRetries = createScheduledRetryPool();

      onClientClose = createChainedCloseHandler(
          onClientClose, createExecutorCloseHandler(scheduledRetries));
    }

    RetryOptions unaryCallRetryOptions = channelOptions.getUnaryCallRetryOptions();
    channel = new UnaryCallRetryInterceptor(
        channel,
        scheduledRetries,
        METHODS_TO_RETRY_MAP,
        unaryCallRetryOptions.getInitialBackoffMillis(),
        unaryCallRetryOptions.getBackoffMultiplier(),
        unaryCallRetryOptions.getMaxElaspedBackoffMillis());
  }

  if (!Strings.isNullOrEmpty(channelOptions.getCallStatusReportPath())) {
    CallCompletionStatusInterceptor postRetryCallStatusInterceptor =
        new CallCompletionStatusInterceptor();

    registerCallStatusReportingShutdownHook(
        channelOptions.getCallStatusReportPath(),
        preRetryCallStatusInterceptor,
        postRetryCallStatusInterceptor);

    channel = ClientInterceptors.intercept(channel, postRetryCallStatusInterceptor);
  }

  return createCloseableChannel(channel, onClientClose);
}
 

import io.grpc.ClientInterceptor; //导入依赖的package包/类
public static ClientInterceptor attachMetadataFromContextInterceptor() {
  return MetadataUtils.newAttachHeadersInterceptor(headersFromCurrentContext());
}