Java 类io.grpc.ClientInterceptor 实例源码

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

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

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

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

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

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

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

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

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

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

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

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

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

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

public static ClientInterceptor instance() {
  return new HeaderClientInterceptor();
}

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

}

public GrpcClientInitializer(ClientOptions clientOptions, List<ClientInterceptor> clientInterceptosr) {
  this(clientOptions, clientInterceptosr, CHANNEL_COUNT_DEFAULT, CHANNEL_COUNT_DEFAULT);
}

/**
 * <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 InterceptorChannel(Channel channel, ClientInterceptor interceptor) {
  this.channel = channel;
  this.interceptor = Preconditions.checkNotNull(interceptor, "interceptor");
}

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

public static ClientInterceptor attachMetadataFromContextInterceptor() {
  return MetadataUtils.newAttachHeadersInterceptor(headersFromCurrentContext());
}

@Override
public final T intercept(List<ClientInterceptor> interceptors) {
  this.interceptors.addAll(interceptors);
  return thisT();
}

@Override
public final T intercept(ClientInterceptor... interceptors) {
  return intercept(Arrays.asList(interceptors));
}

@Override
public ClientInterceptor getClientInterceptor(String fullMethodName) {
  return null;
}

/**
 * Returns the client interceptor that facilitates Census-based stats reporting.
 */
ClientInterceptor getClientInterceptor(boolean recordStartedRpcs, boolean recordFinishedRpcs) {
  return new StatsClientInterceptor(recordStartedRpcs, recordFinishedRpcs);
}

ManagedChannelImpl(
    AbstractManagedChannelImplBuilder<?> builder,
    ClientTransportFactory clientTransportFactory,
    BackoffPolicy.Provider backoffPolicyProvider,
    ObjectPool<? extends Executor> oobExecutorPool,
    Supplier<Stopwatch> stopwatchSupplier,
    List<ClientInterceptor> interceptors,
    ProxyDetector proxyDetector,
    CallTracer.Factory callTracerFactory) {
  this.target = checkNotNull(builder.target, "target");
  this.nameResolverFactory = builder.getNameResolverFactory();
  this.nameResolverParams = checkNotNull(builder.getNameResolverParams(), "nameResolverParams");
  this.nameResolver = getNameResolver(target, nameResolverFactory, nameResolverParams);
  this.loadBalancerFactory =
      checkNotNull(builder.loadBalancerFactory, "loadBalancerFactory");
  this.executorPool = checkNotNull(builder.executorPool, "executorPool");
  this.oobExecutorPool = checkNotNull(oobExecutorPool, "oobExecutorPool");
  this.executor = checkNotNull(executorPool.getObject(), "executor");
  this.delayedTransport = new DelayedClientTransport(this.executor, this.channelExecutor);
  this.delayedTransport.start(delayedTransportListener);
  this.backoffPolicyProvider = backoffPolicyProvider;
  this.transportFactory =
      new CallCredentialsApplyingTransportFactory(clientTransportFactory, this.executor);
  Channel channel = new RealChannel();
  if (builder.binlogProvider != null) {
    channel = builder.binlogProvider.wrapChannel(channel);
  }
  this.interceptorChannel = ClientInterceptors.intercept(channel, interceptors);
  this.stopwatchSupplier = checkNotNull(stopwatchSupplier, "stopwatchSupplier");
  if (builder.idleTimeoutMillis == IDLE_TIMEOUT_MILLIS_DISABLE) {
    this.idleTimeoutMillis = builder.idleTimeoutMillis;
  } else {
    checkArgument(
        builder.idleTimeoutMillis
            >= AbstractManagedChannelImplBuilder.IDLE_MODE_MIN_TIMEOUT_MILLIS,
        "invalid idleTimeoutMillis %s", builder.idleTimeoutMillis);
    this.idleTimeoutMillis = builder.idleTimeoutMillis;
  }
  this.fullStreamDecompression = builder.fullStreamDecompression;
  this.decompressorRegistry = checkNotNull(builder.decompressorRegistry, "decompressorRegistry");
  this.compressorRegistry = checkNotNull(builder.compressorRegistry, "compressorRegistry");
  this.userAgent = builder.userAgent;
  this.proxyDetector = proxyDetector;

  this.channelBufferLimit = builder.retryBufferSize;
  this.perRpcBufferLimit = builder.perRpcBufferLimit;

  phantom = new ManagedChannelReference(this);
  this.callTracerFactory = callTracerFactory;
  channelCallTracer = callTracerFactory.create();
  logger.log(Level.FINE, "[{0}] Created with target {1}", new Object[] {getLogId(), target});
}

/**
 * Returns the client interceptor that facilitates Census-based stats reporting.
 */
ClientInterceptor getClientInterceptor() {
  return clientInterceptor;
}

private void createChannel(
    NameResolver.Factory nameResolverFactory, List<ClientInterceptor> interceptors) {
  createChannel(
      nameResolverFactory, interceptors, true /* requestConnection */,
      ManagedChannelImpl.IDLE_TIMEOUT_MILLIS_DISABLE);
}

private void createChannel(
    NameResolver.Factory nameResolverFactory, List<ClientInterceptor> interceptors,
    boolean requestConnection, long idleTimeoutMillis) {
  class Builder extends AbstractManagedChannelImplBuilder<Builder> {
    Builder(String target) {
      super(target);
    }

    @Override protected ClientTransportFactory buildTransportFactory() {
      throw new UnsupportedOperationException();
    }

    @Override protected Attributes getNameResolverParams() {
      return NAME_RESOLVER_PARAMS;
    }

    @Override public Builder usePlaintext(boolean b) {
      throw new UnsupportedOperationException();
    }
  }

  Builder builder = new Builder(target)
      .nameResolverFactory(nameResolverFactory)
      .loadBalancerFactory(mockLoadBalancerFactory)
      .userAgent(userAgent);
  builder.executorPool = executorPool;
  builder.idleTimeoutMillis = idleTimeoutMillis;
  builder.binlogProvider = binlogProvider;
  checkState(channel == null);
  channel = new ManagedChannelImpl(
      builder, mockTransportFactory, new FakeBackoffPolicyProvider(),
      oobExecutorPool, timer.getStopwatchSupplier(), interceptors, GrpcUtil.NOOP_PROXY_DETECTOR,
      channelStatsFactory);

  if (requestConnection) {
    // Force-exit the initial idle-mode
    channel.exitIdleMode();
    assertEquals(
        idleTimeoutMillis == ManagedChannelImpl.IDLE_TIMEOUT_MILLIS_DISABLE ? 0 : 1,
        timer.numPendingTasks());

    ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
    verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
    helper = helperCaptor.getValue();
  }
}

@Test
public void binaryLogTest() throws Exception {
  final List<Object> capturedReqs = new ArrayList<Object>();
  final class TracingClientInterceptor implements ClientInterceptor {
    private final List<MethodDescriptor<?, ?>> interceptedMethods =
        new ArrayList<MethodDescriptor<?, ?>>();

    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
      interceptedMethods.add(method);
      return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
        @Override
        public void sendMessage(ReqT message) {
          capturedReqs.add(message);
          super.sendMessage(message);
        }
      };
    }
  }

  TracingClientInterceptor userInterceptor = new TracingClientInterceptor();
  binlogProvider = new BinaryLogProvider() {
    @Nullable
    @Override
    public ServerInterceptor getServerInterceptor(String fullMethodName) {
      return null;
    }

    @Override
    public ClientInterceptor getClientInterceptor(String fullMethodName) {
      return new TracingClientInterceptor();
    }

    @Override
    protected int priority() {
      return 0;
    }
  };
  createChannel(
      new FakeNameResolverFactory(true),
      Collections.<ClientInterceptor>singletonList(userInterceptor));
  ClientCall<String, Integer> call =
      channel.newCall(method, CallOptions.DEFAULT.withDeadlineAfter(0, TimeUnit.NANOSECONDS));
  ClientCall.Listener<Integer> listener = new NoopClientCallListener<Integer>();
  call.start(listener, new Metadata());
  assertEquals(1, executor.runDueTasks());

  String actualRequest = "hello world";
  call.sendMessage(actualRequest);

  // The user supplied interceptor must still operate on the original message types
  assertThat(userInterceptor.interceptedMethods).hasSize(1);
  assertSame(
      method.getRequestMarshaller(),
      userInterceptor.interceptedMethods.get(0).getRequestMarshaller());
  assertSame(
      method.getResponseMarshaller(),
      userInterceptor.interceptedMethods.get(0).getResponseMarshaller());

  // The binlog interceptor must be closest to the transport
  assertThat(capturedReqs).hasSize(2);
  // The InputStream is already spent, so just check its type rather than contents
  assertEquals(actualRequest, capturedReqs.get(0));
  assertThat(capturedReqs.get(1)).isInstanceOf(InputStream.class);
}

@Before
public void setUp() {
  MockitoAnnotations.initMocks(this);
  when(mockLoadBalancerFactory.newLoadBalancer(any(Helper.class))).thenReturn(mockLoadBalancer);
  when(mockNameResolver.getServiceAuthority()).thenReturn(AUTHORITY);
  when(mockNameResolverFactory
      .newNameResolver(any(URI.class), any(Attributes.class)))
      .thenReturn(mockNameResolver);
  when(mockTransportFactory.getScheduledExecutorService())
      .thenReturn(timer.getScheduledExecutorService());

  class Builder extends AbstractManagedChannelImplBuilder<Builder> {
    Builder(String target) {
      super(target);
    }

    @Override protected ClientTransportFactory buildTransportFactory() {
      throw new UnsupportedOperationException();
    }

    @Override public Builder usePlaintext(boolean b) {
      throw new UnsupportedOperationException();
    }
  }

  Builder builder = new Builder("fake://target")
      .nameResolverFactory(mockNameResolverFactory)
      .loadBalancerFactory(mockLoadBalancerFactory)
      .idleTimeout(IDLE_TIMEOUT_SECONDS, TimeUnit.SECONDS)
      .userAgent(USER_AGENT);
  builder.executorPool = executorPool;
  channel = new ManagedChannelImpl(
      builder, mockTransportFactory, new FakeBackoffPolicyProvider(),
      oobExecutorPool, timer.getStopwatchSupplier(),
      Collections.<ClientInterceptor>emptyList(),
      GrpcUtil.NOOP_PROXY_DETECTOR, CallTracer.getDefaultFactory());
  newTransports = TestUtils.captureTransports(mockTransportFactory);

  for (int i = 0; i < 2; i++) {
    ArrayList<SocketAddress> addrs = Lists.newArrayList();
    for (int j = 0; j < 2; j++) {
      addrs.add(new FakeSocketAddress("servergroup" + i + "server" + j));
    }
    servers.add(new EquivalentAddressGroup(addrs));
  }
  verify(mockNameResolverFactory).newNameResolver(any(URI.class), any(Attributes.class));
  // Verify the initial idleness
  verify(mockLoadBalancerFactory, never()).newLoadBalancer(any(Helper.class));
  verify(mockTransportFactory, never()).newClientTransport(
      any(SocketAddress.class), anyString(), anyString(), any(ProxyParameters.class));
  verify(mockNameResolver, never()).start(any(NameResolver.Listener.class));
}

@Override
public ClientInterceptor getClientInterceptor(String fullMethodName) {
  return clientBinlogInterceptor;
}

@Nullable
@Override
public ClientInterceptor getClientInterceptor(String fullMethodName) {
  throw new UnsupportedOperationException();
}

/**
 * Create a new {@link Channel} that will call {@code interceptors} before starting a call on the
 * given channel. The first interceptor will have its {@link ClientInterceptor#interceptCall}
 * called first.
 *
 * @param channel the underlying channel to intercept.
 * @param interceptors a list of interceptors to bind to {@code channel}.
 * @return a new channel instance with the interceptors applied.
 */
public static Channel interceptForward(Channel channel,
    List<? extends ClientInterceptor> interceptors) {
  List<? extends ClientInterceptor> copy = new ArrayList<ClientInterceptor>(interceptors);
  Collections.reverse(copy);
  return intercept(channel, copy);
}

/**
 * Create a new {@link Channel} that will call {@code interceptors} before starting a call on the
 * given channel. The last interceptor will have its {@link ClientInterceptor#interceptCall}
 * called first.
 *
 * @param channel the underlying channel to intercept.
 * @param interceptors a list of interceptors to bind to {@code channel}.
 * @return a new channel instance with the interceptors applied.
 */
public static Channel intercept(Channel channel, List<? extends ClientInterceptor> interceptors) {
  Preconditions.checkNotNull(channel);
  for (ClientInterceptor interceptor : interceptors) {
    channel = new InterceptorChannel(channel, interceptor);
  }
  return channel;
}

/**
 * Create a new {@link Channel} that will call {@code interceptors} before starting a call on the
 * given channel. The first interceptor will have its {@link ClientInterceptor#interceptCall}
 * called first.
 *
 * @param channel the underlying channel to intercept.
 * @param interceptors array of interceptors to bind to {@code channel}.
 * @return a new channel instance with the interceptors applied.
 */
public static Channel interceptForward(Channel channel, ClientInterceptor... interceptors) {
  return interceptForward(channel, Arrays.asList(interceptors));
}

/**
 * Create a new {@link Channel} that will call {@code interceptors} before starting a call on the
 * given channel. The last interceptor will have its {@link ClientInterceptor#interceptCall}
 * called first.
 *
 * @param channel the underlying channel to intercept.
 * @param interceptors array of interceptors to bind to {@code channel}.
 * @return a new channel instance with the interceptors applied.
 */
public static Channel intercept(Channel channel, ClientInterceptor... interceptors) {
  return intercept(channel, Arrays.asList(interceptors));
}

/**
 * Returns a {@link ClientInterceptor} for binary logging. gRPC is free to cache the interceptor,
 * so the interceptor must be reusable across calls. At runtime, the request and response
 * marshallers are always {@code Marshaller<InputStream>}.
 * Returns {@code null} if this method is not binary logged.
 */
// TODO(zpencer): ensure the interceptor properly handles retries and hedging
@Nullable
public abstract ClientInterceptor getClientInterceptor(String fullMethodName);

/**
 * Returns a client interceptor that attaches a set of headers to requests.
 *
 * @param extraHeaders the headers to be passed by each call that is processed by the returned
 *                     interceptor
 */
public static ClientInterceptor newAttachHeadersInterceptor(Metadata extraHeaders) {
  return new HeaderAttachingClientInterceptor(extraHeaders);
}