Java 类io.grpc.CallOptions 实例源码

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

@Test
public void noOverwriteWorksCustomKeys() {
    CallOptions.Key<String> k1 = CallOptions.Key.of("k1", null);
    CallOptions.Key<String> k2 = CallOptions.Key.of("k2", null);
    CallOptions.Key<String> k3 = CallOptions.Key.of("k3", null);

    CallOptions baseOptions = CallOptions.DEFAULT.withOption(k1, "FOO").withOption(k3, "BAZ");
    CallOptions defaultOptions = CallOptions.DEFAULT.withOption(k2, "BAR").withOption(k3, "BOP");

    DefaultCallOptionsClientInterceptor interceptor = new DefaultCallOptionsClientInterceptor(defaultOptions);

    CallOptions patchedOptions = interceptor.patchOptions(baseOptions);

    assertThat(patchedOptions.getOption(k1)).isEqualTo("FOO");
    assertThat(patchedOptions.getOption(k2)).isEqualTo("BAR");
    assertThat(patchedOptions.getOption(k3)).isEqualTo("BAZ");
}

@Test
public void overwriteWorksCustomKeys() {
    CallOptions.Key<String> k1 = CallOptions.Key.of("k1", null);
    CallOptions.Key<String> k2 = CallOptions.Key.of("k2", null);
    CallOptions.Key<String> k3 = CallOptions.Key.of("k3", null);

    CallOptions baseOptions = CallOptions.DEFAULT.withOption(k1, "FOO").withOption(k3, "BAZ");
    CallOptions defaultOptions = CallOptions.DEFAULT.withOption(k2, "BAR").withOption(k3, "BOP");

    DefaultCallOptionsClientInterceptor interceptor = new DefaultCallOptionsClientInterceptor(defaultOptions)
            .overwriteExistingValues();

    CallOptions patchedOptions = interceptor.patchOptions(baseOptions);

    assertThat(patchedOptions.getOption(k1)).isEqualTo("FOO");
    assertThat(patchedOptions.getOption(k2)).isEqualTo("BAR");
    assertThat(patchedOptions.getOption(k3)).isEqualTo("BOP");
}

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

@Test
public void testFlowControl() {
  CallOptions callOptions = CallOptions.DEFAULT
      .withOption(GrpcRxOptions.LOW_WATERMARK, 2)
      .withOption(GrpcRxOptions.HIGH_WATERMARK, 6);
  EchoGrpcRx.EchoStub stub = EchoGrpcRx.newStub(channel, callOptions);

  Flowable<EchoReq> requests = Flowable.range(0, 64)
      .map(i -> {
        logger.info("Emitting request {}", i);

        return EchoReq.newBuilder()
            .setId(i)
            .setValue(bigStr)
            .build();
      });

  TestSubscriber<EchoResp> testSubscriber = new AutoTestSubscriber<>(4);
  stub.bidiStreaming(requests)
      .subscribe(testSubscriber);

  testSubscriber.awaitDone(1, TimeUnit.MINUTES);
  testSubscriber.assertComplete();
}

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

@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 requestWithNoCacheOptionSkipsCache() {
  HelloReply reply1 =
      ClientCalls.blockingUnaryCall(
          channelToUse, safeGreeterSayHelloMethod, CallOptions.DEFAULT, message);
  HelloReply reply2 =
      ClientCalls.blockingUnaryCall(
          channelToUse,
          safeGreeterSayHelloMethod,
          CallOptions.DEFAULT.withOption(SafeMethodCachingInterceptor.NO_CACHE_CALL_OPTION, true),
          message);
  HelloReply reply3 =
      ClientCalls.blockingUnaryCall(
          channelToUse, safeGreeterSayHelloMethod, CallOptions.DEFAULT, message);

  assertNotEquals(reply1, reply2);
  assertSame(reply1, reply3);
}

@Test
public void requestWithOnlyIfCachedOption_unavailableIfNotInCache() {
  try {
    ClientCalls.blockingUnaryCall(
        channelToUse,
        safeGreeterSayHelloMethod,
        CallOptions.DEFAULT.withOption(
            SafeMethodCachingInterceptor.ONLY_IF_CACHED_CALL_OPTION, true),
        message);
    fail("Expected call to fail");
  } catch (StatusRuntimeException sre) {
    assertEquals(Status.UNAVAILABLE.getCode(), sre.getStatus().getCode());
    assertEquals(
        "Unsatisfiable Request (only-if-cached set, but value not in cache)",
        sre.getStatus().getDescription());
  }
}

@Override
public synchronized ClientStream newStream(
    final MethodDescriptor<?, ?> method, final Metadata headers, final CallOptions callOptions) {
  if (shutdownStatus != null) {
    final Status capturedStatus = shutdownStatus;
    final StatsTraceContext statsTraceCtx =
        StatsTraceContext.newClientContext(callOptions, headers);
    return new NoopClientStream() {
      @Override
      public void start(ClientStreamListener listener) {
        statsTraceCtx.clientOutboundHeaders();
        statsTraceCtx.streamClosed(capturedStatus);
        listener.closed(capturedStatus, new Metadata());
      }
    };
  }
  headers.put(GrpcUtil.USER_AGENT_KEY, userAgent);
  return new InProcessStream(method, headers, callOptions, authority).clientStream;
}

@Test
public void separateResponseCacheControlDirectives_parsesWithoutError() throws Exception {
  cacheControlDirectives.add("max-age=1");
  cacheControlDirectives.add("no-store , no-cache");

  HelloReply reply1 =
      ClientCalls.blockingUnaryCall(
          channelToUse, safeGreeterSayHelloMethod, CallOptions.DEFAULT, message);
  HelloReply reply2 =
      ClientCalls.blockingUnaryCall(
          channelToUse, safeGreeterSayHelloMethod, CallOptions.DEFAULT, message);

  assertNotEquals(reply1, reply2);
  Truth.assertThat(cache.internalCache).isEmpty();
  Truth.assertThat(cache.removedKeys).isEmpty();
}

@Test
public void afterResponseMaxAge_cacheEntryInvalidated() throws Exception {
  cacheControlDirectives.add("max-age=1");

  HelloReply reply1 =
      ClientCalls.blockingUnaryCall(
          channelToUse, safeGreeterSayHelloMethod, CallOptions.DEFAULT, message);
  HelloReply reply2 =
      ClientCalls.blockingUnaryCall(
          channelToUse, safeGreeterSayHelloMethod, CallOptions.DEFAULT, message);
  assertSame(reply1, reply2);

  // Wait for cache entry to expire
  sleepAtLeast(1001);

  assertNotEquals(
      reply1,
      ClientCalls.blockingUnaryCall(
          channelToUse, safeGreeterSayHelloMethod, CallOptions.DEFAULT, message));
  Truth.assertThat(cache.removedKeys).hasSize(1);
  assertEquals(
      new SafeMethodCachingInterceptor.Key(
          GreeterGrpc.getSayHelloMethod().getFullMethodName(), message),
      cache.removedKeys.get(0));
}

@Test
public void cacheHit_doesNotResetExpiration() throws Exception {
  cacheControlDirectives.add("max-age=1");

  HelloReply reply1 =
      ClientCalls.blockingUnaryCall(
          channelToUse, safeGreeterSayHelloMethod, CallOptions.DEFAULT, message);
  HelloReply reply2 =
      ClientCalls.blockingUnaryCall(
          channelToUse, safeGreeterSayHelloMethod, CallOptions.DEFAULT, message);

  sleepAtLeast(1001);

  HelloReply reply3 =
      ClientCalls.blockingUnaryCall(
          channelToUse, safeGreeterSayHelloMethod, CallOptions.DEFAULT, message);

  assertSame(reply1, reply2);
  assertNotEquals(reply1, reply3);
  Truth.assertThat(cache.internalCache).hasSize(1);
  Truth.assertThat(cache.removedKeys).hasSize(1);
}

@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 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 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 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 invalidInboundTrailersPropagateToMetadata() throws Exception {
  initTransport();
  MockStreamListener listener = new MockStreamListener();
  OkHttpClientStream stream =
      clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
  stream.start(listener);
  stream.request(1);
  assertContainStream(3);
  // Headers block with EOS without correct content type or status
  frameHandler().headers(true, true, 3, 0, Arrays.asList(new Header("random", "4")),
      HeadersMode.HTTP_20_HEADERS);
  assertNull(listener.headers);
  assertEquals(Status.INTERNAL.getCode(), listener.status.getCode());
  assertNotNull(listener.trailers);
  assertEquals("4", listener.trailers
      .get(Metadata.Key.of("random", Metadata.ASCII_STRING_MARSHALLER)));
  shutdownAndVerify();
}

@Test
public void startAddsMaxSize() {
  CallOptions callOptions =
      baseCallOptions.withMaxInboundMessageSize(1).withMaxOutboundMessageSize(2);
  ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>(
      method,
      new SerializingExecutor(Executors.newSingleThreadExecutor()),
      callOptions,
      provider,
      deadlineCancellationExecutor,
      channelCallTracer)
          .setDecompressorRegistry(decompressorRegistry);

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

  verify(stream).setMaxInboundMessageSize(1);
  verify(stream).setMaxOutboundMessageSize(2);
}

@Test
public void overrideDefaultUserAgent() throws Exception {
  startTransport(3, null, true, DEFAULT_MAX_MESSAGE_SIZE, "fakeUserAgent");
  MockStreamListener listener = new MockStreamListener();
  OkHttpClientStream stream =
      clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
  stream.start(listener);
  List<Header> expectedHeaders = Arrays.asList(SCHEME_HEADER, METHOD_HEADER,
      new Header(Header.TARGET_AUTHORITY, "notarealauthority:80"),
      new Header(Header.TARGET_PATH, "/" + method.getFullMethodName()),
      new Header(GrpcUtil.USER_AGENT_KEY.name(),
          GrpcUtil.getGrpcUserAgent("okhttp", "fakeUserAgent")),
      CONTENT_TYPE_HEADER, TE_HEADER);
  verify(frameWriter, timeout(TIME_OUT_MS))
      .synStream(eq(false), eq(false), eq(3), eq(0), eq(expectedHeaders));
  getStream(3).cancel(Status.CANCELLED);
  shutdownAndVerify();
}

@Override
public ClientStreamTracer newClientStreamTracer(CallOptions callOptions, Metadata headers) {
  ClientTracer tracer = new ClientTracer();
  // TODO(zhangkun83): Once retry or hedging is implemented, a ClientCall may start more than
  // one streams.  We will need to update this file to support them.
  if (streamTracerUpdater != null) {
    checkState(
        streamTracerUpdater.compareAndSet(this, null, tracer),
        "Are you creating multiple streams per call? This class doesn't yet support this case");
  } else {
    checkState(
        streamTracer == null,
        "Are you creating multiple streams per call? This class doesn't yet support this case");
    streamTracer = tracer;
  }
  if (module.propagateTags) {
    headers.discardAll(module.statsHeader);
    if (!module.tagger.empty().equals(parentCtx)) {
      headers.put(module.statsHeader, parentCtx);
    }
  }
  return tracer;
}

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

@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 receivingWindowExceeded() throws Exception {
  initTransport();
  MockStreamListener listener = new MockStreamListener();
  OkHttpClientStream stream =
      clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
  stream.start(listener);
  stream.request(1);

  frameHandler().headers(false, false, 3, 0, grpcResponseHeaders(), HeadersMode.HTTP_20_HEADERS);

  int messageLength = Utils.DEFAULT_WINDOW_SIZE + 1;
  byte[] fakeMessage = new byte[messageLength];
  Buffer buffer = createMessageFrame(fakeMessage);
  int messageFrameLength = (int) buffer.size();
  frameHandler().data(false, 3, buffer, messageFrameLength);

  listener.waitUntilStreamClosed();
  assertEquals(Status.INTERNAL.getCode(), listener.status.getCode());
  assertEquals("Received data size exceeded our receiving window size",
      listener.status.getDescription());
  verify(frameWriter, timeout(TIME_OUT_MS)).rstStream(eq(3), eq(ErrorCode.FLOW_CONTROL_ERROR));
  shutdownAndVerify();
}

@Test
public void receiveLongEnoughDataWithoutHeaderAndTrailer() throws Exception {
  initTransport();
  MockStreamListener listener = new MockStreamListener();
  OkHttpClientStream stream =
      clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
  stream.start(listener);
  stream.request(1);
  Buffer buffer = createMessageFrame(new byte[1000]);
  frameHandler().data(false, 3, buffer, (int) buffer.size());

  // Once we receive enough detail, we cancel the stream. so we should have sent cancel.
  verify(frameWriter, timeout(TIME_OUT_MS)).rstStream(eq(3), eq(ErrorCode.CANCEL));

  listener.waitUntilStreamClosed();
  assertEquals(Status.INTERNAL.getCode(), listener.status.getCode());
  assertTrue(listener.status.getDescription().startsWith("headers not received before payload"));
  assertEquals(0, listener.messages.size());
  shutdownAndVerify();
}

@Test
public void writeBeforeConnected() throws Exception {
  initTransportAndDelayConnected();
  final String message = "Hello Server";
  MockStreamListener listener = new MockStreamListener();
  OkHttpClientStream stream =
      clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
  stream.start(listener);
  InputStream input = new ByteArrayInputStream(message.getBytes(UTF_8));
  stream.writeMessage(input);
  stream.flush();
  // The message should be queued.
  verifyNoMoreInteractions(frameWriter);

  allowTransportConnected();

  // The queued message should be sent out.
  ArgumentCaptor<Buffer> captor = ArgumentCaptor.forClass(Buffer.class);
  verify(frameWriter, timeout(TIME_OUT_MS))
      .data(eq(false), eq(3), captor.capture(), eq(12 + HEADER_LENGTH));
  Buffer sentFrame = captor.getValue();
  assertEquals(createMessageFrame(message), sentFrame);
  stream.cancel(Status.CANCELLED);
  shutdownAndVerify();
}

@Test
public void cancelBeforeConnected() throws Exception {
  initTransportAndDelayConnected();
  final String message = "Hello Server";
  MockStreamListener listener = new MockStreamListener();
  OkHttpClientStream stream =
      clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
  stream.start(listener);
  InputStream input = new ByteArrayInputStream(message.getBytes(UTF_8));
  stream.writeMessage(input);
  stream.flush();
  stream.cancel(Status.CANCELLED);
  verifyNoMoreInteractions(frameWriter);

  allowTransportConnected();
  verifyNoMoreInteractions(frameWriter);
  shutdownAndVerify();
}

@Test
public void traceHeadersPropagateSpanContext() throws Exception {
  CensusTracingModule.ClientCallTracer callTracer =
      censusTracing.newClientCallTracer(fakeClientParentSpan, method);
  Metadata headers = new Metadata();
  callTracer.newClientStreamTracer(CallOptions.DEFAULT, headers);

  verify(mockTracingPropagationHandler).toByteArray(same(fakeClientSpanContext));
  verifyNoMoreInteractions(mockTracingPropagationHandler);
  verify(tracer).spanBuilderWithExplicitParent(
      eq("Sent.package1.service2.method3"), same(fakeClientParentSpan));
  verify(spyClientSpanBuilder).setRecordEvents(eq(true));
  verifyNoMoreInteractions(tracer);
  assertTrue(headers.containsKey(censusTracing.tracingHeader));

  ServerStreamTracer serverTracer =
      censusTracing.getServerTracerFactory().newServerStreamTracer(
          method.getFullMethodName(), headers);
  verify(mockTracingPropagationHandler).fromByteArray(same(binarySpanContext));
  verify(tracer).spanBuilderWithRemoteParent(
      eq("Recv.package1.service2.method3"), same(spyClientSpan.getContext()));
  verify(spyServerSpanBuilder).setRecordEvents(eq(true));

  Context filteredContext = serverTracer.filterContext(Context.ROOT);
  assertSame(spyServerSpan, ContextUtils.CONTEXT_SPAN_KEY.get(filteredContext));
}

@Test
public void newStream_afterTermination() throws Exception {
  // We expect the same general behavior as duringShutdown, but for some transports (e.g., Netty)
  // dealing with afterTermination is harder than duringShutdown.
  server.start(serverListener);
  client = newClientTransport(server);
  runIfNotNull(client.start(mockClientTransportListener));
  verify(mockClientTransportListener, timeout(TIMEOUT_MS)).transportReady();
  Status shutdownReason = Status.UNAVAILABLE.withDescription("shutdown called");
  client.shutdown(shutdownReason);
  verify(mockClientTransportListener, timeout(TIMEOUT_MS)).transportTerminated();
  Thread.sleep(100);
  ClientStream stream = client.newStream(methodDescriptor, new Metadata(), callOptions);
  ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
  stream.start(clientStreamListener);
  assertEquals(
      shutdownReason, clientStreamListener.status.get(TIMEOUT_MS, TimeUnit.MILLISECONDS));
  assertNotNull(clientStreamListener.trailers.get(TIMEOUT_MS, TimeUnit.MILLISECONDS));
  verify(mockClientTransportListener, never()).transportInUse(anyBoolean());
  verify(clientStreamTracerFactory).newClientStreamTracer(
      any(CallOptions.class), any(Metadata.class));
  assertSame(shutdownReason, clientStreamTracer1.getStatus());
  // Assert no interactions
  assertNull(serverStreamTracer1.getServerCallInfo());
}

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