Java 类io.grpc.ClientCall 实例源码

@Override
public Subscription subscribe(@NonNull SubscriptionRequestTO req,
        @NonNull FactObserver observer) {
    SubscriptionImpl<Fact> subscription = SubscriptionImpl.on(observer);

    StreamObserver<FactStoreProto.MSG_Notification> responseObserver = new ClientStreamObserver(
            subscription);

    ClientCall<MSG_SubscriptionRequest, MSG_Notification> call = stub.getChannel().newCall(
            RemoteFactStoreGrpc.METHOD_SUBSCRIBE, stub.getCallOptions()
                    .withWaitForReady()
                    .withCompression("gzip"));

    asyncServerStreamingCall(call, converter.toProto(req), responseObserver);

    return subscription.onClose(() -> {
        cancel(call);
    });
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        final 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 {
            Metadata cachedSaved;
            URI uri = serviceUri(next, method);
            synchronized (this) {
                Map<String, List<String>> latestMetadata = getRequestMetadata(uri);
                if (mLastMetadata == null || mLastMetadata != latestMetadata) {
                    mLastMetadata = latestMetadata;
                    mCached = toHeaders(mLastMetadata);
                }
                cachedSaved = mCached;
            }
            headers.merge(cachedSaved);
            delegate().start(responseListener, headers);
        }
    };
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        final 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 {
            Metadata cachedSaved;
            URI uri = serviceUri(next, method);
            synchronized (this) {
                Map<String, List<String>> latestMetadata = getRequestMetadata(uri);
                if (mLastMetadata == null || mLastMetadata != latestMetadata) {
                    mLastMetadata = latestMetadata;
                    mCached = toHeaders(mLastMetadata);
                }
                cachedSaved = mCached;
            }
            headers.merge(cachedSaved);
            delegate().start(responseListener, headers);
        }
    };
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(final 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 {

            Metadata cachedSaved;
            URI uri = serviceUri(next, method);
            synchronized (GoogleCredentialsInterceptor.this) {
                Map<String, List<String>> latestMetadata = getRequestMetadata(uri);
                if (mLastMetadata == null || mLastMetadata != latestMetadata) {
                    mLastMetadata = latestMetadata;
                    mCached = toHeaders(mLastMetadata);
                }
                cachedSaved = mCached;
            }
            headers.merge(cachedSaved);
            delegate().start(responseListener, headers);
        }
    };
}

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

@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) {
      getToken(next).ifPresent(t -> headers.put(TOKEN, t));
      super.start(new SimpleForwardingClientCallListener<RespT>(responseListener) {
        @Override
        public void onClose(Status status, Metadata trailers) {
          if (isInvalidTokenError(status)) {
            try {
              refreshToken(next);
            } catch (Exception e) {
              // don't throw any error here.
              // rpc will retry on expired auth token.
            }
          }
          super.onClose(status, trailers);
        }
      }, headers);
    }
  };
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        final 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 {
            Metadata cachedSaved;
            URI uri = serviceUri(next, method);
            synchronized (this) {
                Map<String, List<String>> latestMetadata = getRequestMetadata(uri);
                if (mLastMetadata == null || mLastMetadata != latestMetadata) {
                    mLastMetadata = latestMetadata;
                    mCached = toHeaders(mLastMetadata);
                }
                cachedSaved = mCached;
            }
            headers.merge(cachedSaved);
            delegate().start(responseListener, headers);
        }
    };
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT>
    newCall(MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
  final Context timerContext = timer.time();
  final AtomicBoolean decremented = new AtomicBoolean(false);
  return new CheckedForwardingClientCall<ReqT, RespT>(delegate.newCall(methodDescriptor, callOptions)) {
    @Override
    protected void checkedStart(ClientCall.Listener<RespT> responseListener, Metadata headers)
        throws Exception {
      ClientCall.Listener<RespT> timingListener = wrap(responseListener, timerContext, decremented);
      getStats().ACTIVE_RPC_COUNTER.inc();
      getStats().RPC_METER.mark();
      delegate().start(timingListener, headers);
    }

    @Override
    public void cancel(String message, Throwable cause) {
      if (!decremented.getAndSet(true)) {
        getStats().ACTIVE_RPC_COUNTER.dec();
      }
      super.cancel(message, cause);
    }
  };
}

/**
 * Executes a unary call with a response {@link SingleObserver}.
 */
public static <ReqT, RespT> Single<RespT> unaryCall(
    final ClientCall<ReqT, RespT> call,
    final ReqT request) {
  final SingleRequestSender<ReqT> requestSender = new SingleRequestSender<ReqT>(call, request);
  SingleResponseReceiver<RespT> responseReceiver = new SingleResponseReceiver<RespT>(call) {
    @Override
    public void startCall() {
      requestSender.startCall();
      super.startCall();
    }
  };

  call.start(responseReceiver, new Metadata());

  return Single.wrap(responseReceiver.singleSource());
}

/**
 * Executes a server-streaming call with a response {@link Subscriber}.
 */
public static <ReqT, RespT> Flowable<RespT> serverStreamingCall(
    final ClientCall<ReqT, RespT> call,
    ReqT request) {
  final SingleRequestSender<ReqT> requestSender = new SingleRequestSender<ReqT>(call, request);
  StreamingResponseReceiver<RespT> responseReceiver = new StreamingResponseReceiver<RespT>(call) {
    @Override
    public void startCall() {
      requestSender.startCall();
      super.startCall();
    }
  };

  call.start(responseReceiver, new Metadata());

  return Flowable.fromPublisher(responseReceiver.publisher());
}

/**
 * Executes a client-streaming call returning a {@link StreamObserver} for the requestMore messages.
 *
 * @return requestMore stream observer.
 */
public static <ReqT, RespT> Single<RespT> clientStreamingCall(
    ClientCall<ReqT, RespT> call,
    Flowable<ReqT> requests,
    CallOptions options) {
  final StreamRequestSender<ReqT> requestSender = new StreamRequestSender<ReqT>(call, getLowWatermark(options), getHighWatermark(options));
  SingleResponseReceiver<RespT> responseReceiver = new SingleResponseReceiver<RespT>(call) {
    @Override
    public void startCall() {
      requestSender.startCall();
      super.startCall();
    }
  };

  ClientCall.Listener<RespT> delegate = new DelegateClientCallListener<RespT>(requestSender, responseReceiver);
  call.start(delegate, new Metadata());

  requests.subscribe(requestSender.subscriber());

  return Single.wrap(responseReceiver.singleSource());
}

/**
 * Executes a bidi-streaming call.
 *
 * @return requestMore stream observer.
 */
public static <ReqT, RespT> Flowable<RespT> bidiStreamingCall(
    ClientCall<ReqT, RespT> call,
    Flowable<ReqT> requests,
    CallOptions options) {
  final StreamRequestSender<ReqT> requestSender = new StreamRequestSender<ReqT>(call, getLowWatermark(options), getHighWatermark(options));
  StreamingResponseReceiver<RespT> responseReceiver = new StreamingResponseReceiver<RespT>(call) {
    @Override
    public void startCall() {
      requestSender.startCall();
      super.startCall();
    }
  };

  ClientCall.Listener<RespT> delegate = new DelegateClientCallListener<RespT>(requestSender, responseReceiver);
  call.start(delegate, new Metadata());

  requests.subscribe(requestSender.subscriber());

  return Flowable.fromPublisher(responseReceiver.publisher());
}

public SingleResponseReceiver(ClientCall<?, RespT> call) {
  this.call = call;

  this.source = new SingleSource<RespT>() {
    @Override
    public void subscribe(SingleObserver<? super RespT> observer) {
      responseObserver = observer;

      // todo which disposable should be used here
      observer.onSubscribe(Disposables.disposed());

      // start call until response gets subscribed
      startCall();

      if (error != null) {
        responseObserver.onError(error);
        error = null;
      }
    }
  };
}

public StreamRequestSender(ClientCall<ReqT, ?> call, int lowWatermark, int highWatermark) {
  this.call = call;

  grpcSubscriber  = new GrpcSubscriber<ReqT>(lowWatermark, highWatermark) {
    @Override
    protected boolean isReady() {
      return StreamRequestSender.this.call.isReady();
    }

    @Override
    protected void sendMessage(ReqT req) {
      StreamRequestSender.this.call.sendMessage(req);
    }

    @Override
    protected void error(Throwable t) {
      StreamRequestSender.this.call.cancel("Upstream error", t);
    }

    @Override
    protected void complete() {
      StreamRequestSender.this.call.halfClose();
    }
  };
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
    MethodDescriptor<ReqT,RespT> method, CallOptions callOptions, Channel next) {
  LOGGER.info("Intercepted " + method.getFullMethodName());
  ClientCall<ReqT, RespT> call = next.newCall(method, callOptions);

  call = new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(call) {
    @Override
    public void start(Listener<RespT> responseListener, Metadata headers) {
      if (apiKey != null && !apiKey.isEmpty()) {
        LOGGER.info("Attaching API Key: " + apiKey);
        headers.put(API_KEY_HEADER, apiKey);
      }
      super.start(responseListener, headers);
    }
  };
  return call;
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
    MethodDescriptor<ReqT,RespT> method, CallOptions callOptions, Channel next) {
  LOGGER.info("Intercepted " + method.getFullMethodName());
  ClientCall<ReqT, RespT> call = next.newCall(method, callOptions);

  call = new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(call) {
    @Override
    public void start(Listener<RespT> responseListener, Metadata headers) {
      if (apiKey != null && !apiKey.isEmpty()) {
        LOGGER.info("Attaching API Key: " + apiKey);
        headers.put(API_KEY_HEADER, apiKey);
      }
      if (authToken != null && !authToken.isEmpty()) {
        System.out.println("Attaching auth token");
        headers.put(AUTHORIZATION_HEADER, "Bearer " + authToken);
      }
      super.start(responseListener, headers);
    }
  };
  return call;
}

/**
 * This is more advanced and does not make use of the stub.  You should not normally need to do
 * this, but here is how you would.
 */
void advancedAsyncCall() {
  ClientCall<HelloRequest, HelloReply> call =
      channel.newCall(GreeterGrpc.getSayHelloMethod(), CallOptions.DEFAULT);

  final CountDownLatch latch = new CountDownLatch(1);

  call.start(new ClientCall.Listener<HelloReply>() {

    @Override
    public void onClose(Status status, Metadata trailers) {
      Verify.verify(status.getCode() == Status.Code.INTERNAL);
      Verify.verify(status.getDescription().contains("Narwhal"));
      // Cause is not transmitted over the wire.
      latch.countDown();
    }
  }, new Metadata());

  call.sendMessage(HelloRequest.newBuilder().setName("Marge").build());
  call.halfClose();

  if (!Uninterruptibles.awaitUninterruptibly(latch, 1, TimeUnit.SECONDS)) {
    throw new RuntimeException("timeout!");
  }
}

@Test
public void testCopyCredentialToHeaders() throws IOException {
  ListMultimap<String, String> values = LinkedListMultimap.create();
  values.put("Authorization", "token1");
  values.put("Authorization", "token2");
  values.put("Extra-Authorization", "token3");
  values.put("Extra-Authorization", "token4");
  when(credentials.getRequestMetadata(any(URI.class))).thenReturn(Multimaps.asMap(values));
  ClientCall<String, Integer> interceptedCall =
      interceptor.interceptCall(descriptor, CallOptions.DEFAULT, channel);
  Metadata headers = new Metadata();
  interceptedCall.start(listener, headers);
  assertEquals(listener, call.responseListener);
  assertEquals(headers, call.headers);

  Iterable<String> authorization = headers.getAll(AUTHORIZATION);
  Assert.assertArrayEquals(new String[]{"token1", "token2"},
      Iterables.toArray(authorization, String.class));
  Iterable<String> extraAuthorization = headers.getAll(EXTRA_AUTHORIZATION);
  Assert.assertArrayEquals(new String[]{"token3", "token4"},
      Iterables.toArray(extraAuthorization, String.class));
}

@Test
public void testWithOAuth2Credential() {
  final AccessToken token = new AccessToken("allyourbase", new Date(Long.MAX_VALUE));
  final OAuth2Credentials oAuth2Credentials = new OAuth2Credentials() {
    @Override
    public AccessToken refreshAccessToken() throws IOException {
      return token;
    }
  };
  interceptor = new ClientAuthInterceptor(oAuth2Credentials, executor);
  ClientCall<String, Integer> interceptedCall =
      interceptor.interceptCall(descriptor, CallOptions.DEFAULT, channel);
  Metadata headers = new Metadata();
  interceptedCall.start(listener, headers);
  assertEquals(listener, call.responseListener);
  assertEquals(headers, call.headers);
  Iterable<String> authorization = headers.getAll(AUTHORIZATION);
  Assert.assertArrayEquals(new String[]{"Bearer allyourbase"},
      Iterables.toArray(authorization, String.class));
}

@Test
public void verifyServiceUri() throws IOException {
  ClientCall<String, Integer> interceptedCall;

  doReturn("example.com:443").when(channel).authority();
  interceptedCall = interceptor.interceptCall(descriptor, CallOptions.DEFAULT, channel);
  interceptedCall.start(listener, new Metadata());
  verify(credentials).getRequestMetadata(URI.create("https://example.com/a.service"));
  interceptedCall.cancel("Cancel for test", null);

  doReturn("example.com:123").when(channel).authority();
  interceptedCall = interceptor.interceptCall(descriptor, CallOptions.DEFAULT, channel);
  interceptedCall.start(listener, new Metadata());
  verify(credentials).getRequestMetadata(URI.create("https://example.com:123/a.service"));
  interceptedCall.cancel("Cancel for test", null);
}

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

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
    MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
  // New RPCs on client-side inherit the tag context from the current Context.
  TagContext parentCtx = tagger.getCurrentTagContext();
  final ClientCallTracer tracerFactory =
      newClientCallTracer(parentCtx, method.getFullMethodName(),
          recordStartedRpcs, recordFinishedRpcs);
  ClientCall<ReqT, RespT> call =
      next.newCall(method, callOptions.withStreamTracerFactory(tracerFactory));
  return new SimpleForwardingClientCall<ReqT, RespT>(call) {
    @Override
    public void start(Listener<RespT> responseListener, Metadata headers) {
      delegate().start(
          new SimpleForwardingClientCallListener<RespT>(responseListener) {
            @Override
            public void onClose(Status status, Metadata trailers) {
              tracerFactory.callEnded(status);
              super.onClose(status, trailers);
            }
          },
          headers);
    }
  };
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
    MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
  // New RPCs on client-side inherit the tracing context from the current Context.
  // Safe usage of the unsafe trace API because CONTEXT_SPAN_KEY.get() returns the same value
  // as Tracer.getCurrentSpan() except when no value available when the return value is null
  // for the direct access and BlankSpan when Tracer API is used.
  final ClientCallTracer tracerFactory = newClientCallTracer(CONTEXT_SPAN_KEY.get(), method);
  ClientCall<ReqT, RespT> call =
      next.newCall(method, callOptions.withStreamTracerFactory(tracerFactory));
  return new SimpleForwardingClientCall<ReqT, RespT>(call) {
    @Override
    public void start(Listener<RespT> responseListener, Metadata headers) {
      delegate().start(
          new SimpleForwardingClientCallListener<RespT>(responseListener) {
            @Override
            public void onClose(io.grpc.Status status, Metadata trailers) {
              tracerFactory.callEnded(status);
              super.onClose(status, trailers);
            }
          },
          headers);
    }
  };
}

@Test
public void channelStat_callEndFail_oob() throws Exception {
  createChannel(new FakeNameResolverFactory(true), NO_INTERCEPTOR);
  OobChannel oob1 = (OobChannel) helper.createOobChannel(addressGroup, "oob1authority");
  ClientCall<String, Integer> call = oob1.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata());
  call.cancel("msg", null);

  assertEquals(0, getStats(channel).callsSucceeded);
  assertEquals(0, getStats(channel).callsFailed);
  oobExecutor.runDueTasks();
  // only oob channel stats updated
  verify(mockCallListener).onClose(any(Status.class), any(Metadata.class));
  assertEquals(0, getStats(oob1).callsSucceeded);
  assertEquals(1, getStats(oob1).callsFailed);
  assertEquals(0, getStats(channel).callsSucceeded);
  assertEquals(0, getStats(channel).callsFailed);
}

@Test
public void delayedTransportHoldsOffIdleness() throws Exception {
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata());
  assertTrue(channel.inUseStateAggregator.isInUse());

  // As long as the delayed transport is in-use (by the pending RPC), the channel won't go idle.
  timer.forwardTime(IDLE_TIMEOUT_SECONDS * 2, TimeUnit.SECONDS);
  assertTrue(channel.inUseStateAggregator.isInUse());

  // Cancelling the only RPC will reset the in-use state.
  assertEquals(0, executor.numPendingTasks());
  call.cancel("In test", null);
  assertEquals(1, executor.runDueTasks());
  assertFalse(channel.inUseStateAggregator.isInUse());
  // And allow the channel to go idle.
  timer.forwardTime(IDLE_TIMEOUT_SECONDS - 1, TimeUnit.SECONDS);
  verify(mockLoadBalancer, never()).shutdown();
  timer.forwardTime(1, TimeUnit.SECONDS);
  verify(mockLoadBalancer).shutdown();
}

@Test
public void updateSubchannelAddresses_newAddressConnects() {
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata()); // Create LB
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  Helper helper = helperCaptor.getValue();
  Subchannel subchannel = helper.createSubchannel(servers.get(0), Attributes.EMPTY);

  subchannel.requestConnection();
  MockClientTransportInfo t0 = newTransports.poll();
  t0.listener.transportReady();

  helper.updateSubchannelAddresses(subchannel, servers.get(1));

  subchannel.requestConnection();
  MockClientTransportInfo t1 = newTransports.poll();
  t1.listener.transportReady();
}

@Test
public void updateSubchannelAddresses_existingAddressDoesNotConnect() {
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata()); // Create LB
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  Helper helper = helperCaptor.getValue();
  Subchannel subchannel = helper.createSubchannel(servers.get(0), Attributes.EMPTY);

  subchannel.requestConnection();
  MockClientTransportInfo t0 = newTransports.poll();
  t0.listener.transportReady();

  List<SocketAddress> changedList = new ArrayList<SocketAddress>(servers.get(0).getAddresses());
  changedList.add(new FakeSocketAddress("aDifferentServer"));
  helper.updateSubchannelAddresses(subchannel, new EquivalentAddressGroup(changedList));

  subchannel.requestConnection();
  assertNull(newTransports.poll());
}

@Test
public void updateOobChannelAddresses_newAddressConnects() {
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata()); // Create LB
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  Helper helper = helperCaptor.getValue();
  ManagedChannel oobChannel = helper.createOobChannel(servers.get(0), "localhost");

  oobChannel.newCall(method, CallOptions.DEFAULT).start(mockCallListener, new Metadata());
  MockClientTransportInfo t0 = newTransports.poll();
  t0.listener.transportReady();

  helper.updateOobChannelAddresses(oobChannel, servers.get(1));

  oobChannel.newCall(method, CallOptions.DEFAULT).start(mockCallListener, new Metadata());
  MockClientTransportInfo t1 = newTransports.poll();
  t1.listener.transportReady();
}

@Test
public void updateOobChannelAddresses_existingAddressDoesNotConnect() {
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata()); // Create LB
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  Helper helper = helperCaptor.getValue();
  ManagedChannel oobChannel = helper.createOobChannel(servers.get(0), "localhost");

  oobChannel.newCall(method, CallOptions.DEFAULT).start(mockCallListener, new Metadata());
  MockClientTransportInfo t0 = newTransports.poll();
  t0.listener.transportReady();

  List<SocketAddress> changedList = new ArrayList<SocketAddress>(servers.get(0).getAddresses());
  changedList.add(new FakeSocketAddress("aDifferentServer"));
  helper.updateOobChannelAddresses(oobChannel, new EquivalentAddressGroup(changedList));

  oobChannel.newCall(method, CallOptions.DEFAULT).start(mockCallListener, new Metadata());
  assertNull(newTransports.poll());
}

@Test
public void wrapChannel_methodDescriptor() throws Exception {
  final AtomicReference<MethodDescriptor<?, ?>> methodRef =
      new AtomicReference<MethodDescriptor<?, ?>>();
  Channel channel = new Channel() {
    @Override
    public <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall(
        MethodDescriptor<RequestT, ResponseT> method, CallOptions callOptions) {
      methodRef.set(method);
      return new NoopClientCall<RequestT, ResponseT>();
    }

    @Override
    public String authority() {
      throw new UnsupportedOperationException();
    }
  };
  Channel wChannel = binlogProvider.wrapChannel(channel);
  ClientCall<String, Integer> ignoredClientCall = wChannel.newCall(method, CallOptions.DEFAULT);
  validateWrappedMethod(methodRef.get());
}

/**
 * Cancels a call, and throws the exception.
 *
 * @param t must be a RuntimeException or Error
 */
private static RuntimeException cancelThrow(ClientCall<?, ?> call, Throwable t) {
  try {
    call.cancel(null, t);
  } catch (Throwable e) {
    assert e instanceof RuntimeException || e instanceof Error;
    logger.log(Level.SEVERE, "RuntimeException encountered while closing call", e);
  }
  if (t instanceof RuntimeException) {
    throw (RuntimeException) t;
  } else if (t instanceof Error) {
    throw (Error) t;
  }
  // should be impossible
  throw new AssertionError(t);
}

@Test
public void unaryBlockingCallSuccess() throws Exception {
  Integer req = 2;
  final String resp = "bar";
  final Status status = Status.OK;
  final Metadata trailers = new Metadata();

  NoopClientCall<Integer, String> call = new NoopClientCall<Integer, String>() {
    @Override
    public void start(ClientCall.Listener<String> listener, Metadata headers) {
      listener.onMessage(resp);
      listener.onClose(status, trailers);
    }
  };

  String actualResponse = ClientCalls.blockingUnaryCall(call, req);
  assertEquals(resp, actualResponse);
}

@Test
public void unaryBlockingCallFailed() throws Exception {
  Integer req = 2;
  final Status status = Status.INTERNAL.withDescription("Unique status");
  final Metadata trailers = new Metadata();

  NoopClientCall<Integer, String> call = new NoopClientCall<Integer, String>() {
    @Override
    public void start(io.grpc.ClientCall.Listener<String> listener, Metadata headers) {
      listener.onClose(status, trailers);
    }
  };

  try {
    ClientCalls.blockingUnaryCall(call, req);
    fail("Should fail");
  } catch (StatusRuntimeException e) {
    assertSame(status, e.getStatus());
    assertSame(trailers, e.getTrailers());
  }
}

@Test
public void unaryFutureCallFailed() throws Exception {
  final AtomicReference<ClientCall.Listener<String>> listener =
      new AtomicReference<ClientCall.Listener<String>>();
  NoopClientCall<Integer, String> call = new NoopClientCall<Integer, String>() {
    @Override
    public void start(io.grpc.ClientCall.Listener<String> responseListener, Metadata headers) {
      listener.set(responseListener);
    }
  };
  Integer req = 2;
  ListenableFuture<String> future = ClientCalls.futureUnaryCall(call, req);
  Metadata trailers = new Metadata();
  listener.get().onClose(Status.INTERNAL, trailers);
  try {
    future.get();
    fail("Should fail");
  } catch (ExecutionException e) {
    Status status = Status.fromThrowable(e);
    assertEquals(Status.INTERNAL, status);
    Metadata metadata = Status.trailersFromThrowable(e);
    assertSame(trailers, metadata);
  }
}

@Test
public void blockingResponseStreamFailed() throws Exception {
  final AtomicReference<ClientCall.Listener<String>> listener =
      new AtomicReference<ClientCall.Listener<String>>();
  NoopClientCall<Integer, String> call = new NoopClientCall<Integer, String>() {
    @Override
    public void start(io.grpc.ClientCall.Listener<String> responseListener, Metadata headers) {
      listener.set(responseListener);
    }
  };

  Integer req = 2;
  Iterator<String> iter = ClientCalls.blockingServerStreamingCall(call, req);

  Metadata trailers = new Metadata();
  listener.get().onClose(Status.INTERNAL, trailers);
  try {
    iter.next();
    fail("Should fail");
  } catch (Exception e) {
    Status status = Status.fromThrowable(e);
    assertEquals(Status.INTERNAL, status);
    Metadata metadata = Status.trailersFromThrowable(e);
    assertSame(trailers, metadata);
  }
}

static void startCall(ClientCall<Message, Message> call,
    ClientCall.Listener<Message> responseListener, boolean streamingResponse) {
  call.start(responseListener, new Metadata());
  if (streamingResponse) {
    call.request(1);
  } else {
    call.request(2);
  }
}

/** Say hello to server. */
public void greet(final String name) {
  final ClientCall<HelloRequest, HelloReply> call =
      channel.newCall(GreeterGrpc.METHOD_SAY_HELLO, CallOptions.DEFAULT);

  final CountDownLatch latch = new CountDownLatch(1);

  call.start(new Listener<HelloReply>() {
    @Override
    public void onHeaders(Metadata headers) {
      super.onHeaders(headers);
      String encoding = headers.get(GrpcUtil.MESSAGE_ENCODING_KEY);
      if (encoding == null) {
        throw new RuntimeException("No compression selected!");
      }
    }

    @Override
    public void onMessage(HelloReply message) {
      super.onMessage(message);
      logger.info("Greeting: " + message.getMessage());
      latch.countDown();
    }

    @Override
    public void onClose(Status status, Metadata trailers) {
      latch.countDown();
      if (!status.isOk()) {
        throw status.asRuntimeException();
      }
    }
  }, new Metadata());

  call.setMessageCompression(true);
  call.sendMessage(HelloRequest.newBuilder().setName(name).build());
  call.request(1);
  call.halfClose();

  Uninterruptibles.awaitUninterruptibly(latch, 100, TimeUnit.SECONDS);
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,
    CallOptions callOptions, Channel next) {
  Integer timeout = callOptions.getOption(DEADLINE_KEY);
  CallOptions callOptionsWithTimeout = null != timeout && timeout > 0 ? 
      callOptions.withDeadlineAfter(timeout, TimeUnit.MILLISECONDS) : callOptions;
  return next.newCall(method, callOptionsWithTimeout);
}

protected <RespT> ClientCall.Listener<RespT> wrap(final ClientCall.Listener<RespT> delegate,
    final Context timeContext, final AtomicBoolean decremented) {
  return new ClientCall.Listener<RespT>() {

    @Override
    public void onHeaders(Metadata headers) {
      delegate.onHeaders(headers);
    }

    @Override
    public void onMessage(RespT message) {
      delegate.onMessage(message);
    }

    @Override
    public void onClose(Status status, Metadata trailers) {
      try {
        if (trailers != null) {
          // Be extra defensive since this is only used for logging
          trailers.put(CHANNEL_ID_KEY, Integer.toString(channelId));
        }
        if (!decremented.getAndSet(true)) {
          getStats().ACTIVE_RPC_COUNTER.dec();
        }
        if (!status.isOk()) {
          ClientMetrics.meter(MetricLevel.Info, "grpc.errors." + status.getCode().name())
              .mark();
        }
        delegate.onClose(status, trailers);
      } finally {
        timeContext.close();
      }
    }

    @Override
    public void onReady() {
      delegate.onReady();
    }
  };
}

@Override
@SuppressWarnings("unchecked")
public final void start(Listener<RespT> responseListener, Metadata headers) {
  try {
    checkedStart(responseListener, headers);
  } catch (Exception e) {
    // Because start() doesn't throw, the caller may still try to call other methods on this
    // call object. Passing these invocations to the original delegate will cause
    // IllegalStateException because delegate().start() was not called. We switch the delegate
    // to a NO-OP one to prevent the IllegalStateException. The user will finally get notified
    // about the error through the listener.
    delegate = (ClientCall<ReqT, RespT>) NOOP_CALL;
    responseListener.onClose(Status.fromThrowable(e), new Metadata());
  }
}