Java 类io.grpc.MethodDescriptor 实例源码

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

protected <ReqT, RespT> RespT callWithRetry(MethodDescriptor<ReqT, RespT> method,
                                            Supplier<ReqT> requestFactory,
                                            ErrorHandler<RespT> handler) {
  if (logger.isTraceEnabled()) {
    logger.trace(String.format("Calling %s...", method.getFullMethodName()));
  }
  RetryPolicy.Builder<RespT> builder = new Builder<>(conf.getRetryTimes(), conf.getBackOffClass());
  RespT resp =
      builder.create(handler)
          .callWithRetry(
              () -> {
                BlockingStubT stub = getBlockingStub();
                return ClientCalls.blockingUnaryCall(
                    stub.getChannel(), method, stub.getCallOptions(), requestFactory.get());
              },
              method.getFullMethodName());
  if (logger.isTraceEnabled()) {
    logger.trace(String.format("leaving %s...", method.getFullMethodName()));
  }
  return resp;
}

protected <ReqT, RespT> void callAsyncWithRetry(
    MethodDescriptor<ReqT, RespT> method,
    Supplier<ReqT> requestFactory,
    StreamObserver<RespT> responseObserver,
    ErrorHandler<RespT> handler) {
  logger.debug(String.format("Calling %s...", method.getFullMethodName()));

  RetryPolicy.Builder<RespT> builder = new Builder<>(conf.getRetryTimes(), conf.getBackOffClass());
  builder.create(handler)
      .callWithRetry(
          () -> {
            StubT stub = getAsyncStub();
            ClientCalls.asyncUnaryCall(
                stub.getChannel().newCall(method, stub.getCallOptions()),
                requestFactory.get(),
                responseObserver);
            return null;
          },
          method.getFullMethodName());
  logger.debug(String.format("leaving %s...", method.getFullMethodName()));
}

<ReqT, RespT> StreamObserver<ReqT> callBidiStreamingWithRetry(
    MethodDescriptor<ReqT, RespT> method,
    StreamObserver<RespT> responseObserver,
    ErrorHandler<StreamObserver<ReqT>> handler) {
  logger.debug(String.format("Calling %s...", method.getFullMethodName()));

  RetryPolicy.Builder<StreamObserver<ReqT>> builder = new Builder<>(conf.getRetryTimes(), conf.getBackOffClass());
  StreamObserver<ReqT> observer =
      builder.create(handler)
          .callWithRetry(
              () -> {
                StubT stub = getAsyncStub();
                return asyncBidiStreamingCall(
                    stub.getChannel().newCall(method, stub.getCallOptions()), responseObserver);
              },
              method.getFullMethodName());
  logger.debug(String.format("leaving %s...", method.getFullMethodName()));
  return observer;
}

protected <ReqT, RespT> StreamingResponse callServerStreamingWithRetry(
    MethodDescriptor<ReqT, RespT> method,
    Supplier<ReqT> requestFactory,
    ErrorHandler<StreamingResponse> handler) {
  logger.debug(String.format("Calling %s...", method.getFullMethodName()));

  RetryPolicy.Builder<StreamingResponse> builder =
      new Builder<>(conf.getRetryTimes(), conf.getBackOffClass());
  StreamingResponse response =
      builder.create(handler)
          .callWithRetry(
              () -> {
                BlockingStubT stub = getBlockingStub();
                return new StreamingResponse(
                    blockingServerStreamingCall(
                        stub.getChannel(),
                        method,
                        stub.getCallOptions(),
                        requestFactory.get()
                    )
                );
              },
              method.getFullMethodName());
  logger.debug(String.format("leaving %s...", method.getFullMethodName()));
  return response;
}

@BeforeClass
public static void startServer() throws IOException {
    AfricasTalking.initialize(Fixtures.USERNAME, Fixtures.API_KEY);
    server = new Server(new Authenticator() {
        @Override
        public boolean authenticate(String client) {
            return client.compareToIgnoreCase(TEST_CLIENT_ID) == 0;
        }
    });
    server.addSipCredentials("test", "secret", "sip://at.dev");
    server.start(certFile, privateKeyFile, TEST_PORT);

    ManagedChannel ch = NettyChannelBuilder.forAddress("localhost", TEST_PORT)
        .sslContext(GrpcSslContexts.forClient().trustManager(certFile).build())
        .build();


    client = SdkServerServiceGrpc.newBlockingStub(ch)
        .withCallCredentials(new CallCredentials(){
            @Override
            public void applyRequestMetadata(MethodDescriptor<?, ?> method, Attributes attrs, Executor appExecutor,
                    final MetadataApplier applier) {
                        appExecutor.execute(new Runnable(){
                            @Override
                            public void run() {
                                try {
                                    Metadata headers = new Metadata();
                                    Metadata.Key<String> clientIdKey = Metadata.Key.of("X-Client-Id", Metadata.ASCII_STRING_MARSHALLER);
                                    headers.put(clientIdKey, TEST_CLIENT_ID);
                                    applier.apply(headers);
                                } catch(Throwable ex) {
                                    applier.fail(Status.UNAUTHENTICATED.withCause(ex));
                                }
                            }
                        });

            }
        });
}

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

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

public UnaryCallRetryInterceptor(
    Channel delegate,
    ScheduledExecutorService executorService,
    Set<MethodDescriptor<?, ?>> retriableMethods,
    int initialBackoffMillis,
    double backoffMultiplier,
    int maxElapsedBackoffMillis) {
  this(
      delegate,
      executorService,
      Maps.asMap(retriableMethods, new Function<MethodDescriptor<?, ?>, Predicate<?>>() {
        @Override
        public Predicate<Object> apply(MethodDescriptor<?, ?> methodDescriptor) {
          return Predicates.alwaysTrue();
        }
      }),
      initialBackoffMillis,
      backoffMultiplier,
      maxElapsedBackoffMillis);
}

@Override
public <ReqT, RespT> Call<ReqT, RespT> newCall(MethodDescriptor<ReqT, RespT> methodDescriptor) {
  if (methodCanBeRetried(methodDescriptor)) {
    ExponentialBackOff.Builder backOffBuilder = new ExponentialBackOff.Builder();
    backOffBuilder.setInitialIntervalMillis(initialBackoffMillis);
    backOffBuilder.setMultiplier(backoffMultiplier);
    backOffBuilder.setMaxElapsedTimeMillis(maxElapsedBackoffMillis);
    Predicate<ReqT> isPayloadRetriablePredicate = getUncheckedPredicate(methodDescriptor);
    return new RetryingCall<>(
        delegate,
        methodDescriptor,
        isPayloadRetriablePredicate,
        executorService,
        backOffBuilder.build());
  }
  return delegate.newCall(methodDescriptor);
}

private static CloseableChannel createCloseableChannel(TransportOptions transportOptions,
    ExecutorService executor){
  final ChannelImpl channelImpl = createChannel(transportOptions, executor);
  return new CloseableChannel() {
    @Override
    public <RequestT, ResponseT> Call<RequestT, ResponseT> newCall(
        MethodDescriptor<RequestT, ResponseT> methodDescriptor) {
      return channelImpl.newCall(methodDescriptor);
    }

    @Override
    public void close() throws IOException {
      channelImpl.shutdown();
      while (!channelImpl.isTerminated()) {
        try {
          channelImpl.awaitTerminated(CHANNEL_TERMINATE_WAIT_SECONDS, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
          Thread.interrupted();
          throw new IOException("Interrupted while sleeping for close", e);
        }
      }
    }
  }; 
}

/**
 * Given a channel, create a CloseableChannel that invokces onClientClose when the close
 * method is invoked.
 */
private static CloseableChannel createCloseableChannel(
    final Channel channel, final ClientCloseHandler onClientClose) {
  return new CloseableChannel() {
    @Override
    public void close() throws IOException {
      onClientClose.close();
    }

    @Override
    public <ReqT, RespT> Call<ReqT, RespT> newCall(
        MethodDescriptor<ReqT, RespT> methodDescriptor) {
      return channel.newCall(methodDescriptor);
    }
  };
}

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

@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
    MethodInterceptResult result) throws Throwable {
    GRPCDynamicFields cachedObjects = (GRPCDynamicFields)objInst.getSkyWalkingDynamicField();
    Metadata headers = cachedObjects.getMetadata();
    Map<String, String> headerMap = new HashMap<String, String>();
    for (String key : headers.keys()) {
        if (!key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
            String value = headers.get(Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER));
            headerMap.put(key, value);
        }
    }

    ContextCarrier contextCarrier = new ContextCarrier();
    CarrierItem next = contextCarrier.items();
    while (next.hasNext()) {
        next = next.next();
        String contextValue = headerMap.get(next.getHeadKey());
        if (!StringUtil.isEmpty(contextValue)) {
            next.setHeadValue(contextValue);
        }
    }

    final AbstractSpan span = ContextManager.createEntrySpan(cachedObjects.getRequestMethodName() + (cachedObjects.getMethodType() != MethodDescriptor.MethodType.UNARY ? STREAM_CALL_OPERATION_NAME_SUFFIX : BLOCK_CALL_OPERATION_NAME_SUFFIX), contextCarrier);
    span.setComponent(ComponentsDefine.GRPC);
}

OkHttpClientStream(
    MethodDescriptor<?, ?> method,
    Metadata headers,
    AsyncFrameWriter frameWriter,
    OkHttpClientTransport transport,
    OutboundFlowController outboundFlow,
    Object lock,
    int maxMessageSize,
    String authority,
    String userAgent,
    StatsTraceContext statsTraceCtx,
    TransportTracer transportTracer) {
  super(
      new OkHttpWritableBufferAllocator(),
      statsTraceCtx,
      transportTracer,
      headers,
      method.isSafe());
  this.statsTraceCtx = checkNotNull(statsTraceCtx, "statsTraceCtx");
  this.method = method;
  this.authority = authority;
  this.userAgent = userAgent;
  this.state = new TransportState(maxMessageSize, statsTraceCtx, lock, frameWriter, outboundFlow,
      transport);
}

@Override
public OkHttpClientStream newStream(final MethodDescriptor<?, ?> method,
    final Metadata headers, CallOptions callOptions) {
  Preconditions.checkNotNull(method, "method");
  Preconditions.checkNotNull(headers, "headers");
  StatsTraceContext statsTraceCtx = StatsTraceContext.newClientContext(callOptions, headers);
  return new OkHttpClientStream(
      method,
      headers,
      frameWriter,
      OkHttpClientTransport.this,
      outboundFlow,
      lock,
      maxMessageSize,
      defaultAuthority,
      userAgent,
      statsTraceCtx,
      transportTracer);
}

@Before
public void setUp() {
  MockitoAnnotations.initMocks(this);
  methodDescriptor = MethodDescriptor.<Void, Void>newBuilder()
      .setType(MethodDescriptor.MethodType.UNARY)
      .setFullMethodName("testService/test")
      .setRequestMarshaller(marshaller)
      .setResponseMarshaller(marshaller)
      .build();

  stream = new OkHttpClientStream(
      methodDescriptor,
      new Metadata(),
      frameWriter,
      transport,
      flowController,
      lock,
      MAX_MESSAGE_SIZE,
      "localhost",
      "userAgent",
      StatsTraceContext.NOOP,
      transportTracer);
}

@Test
public void maxInboundSize_exact() {
  StreamingOutputCallRequest request = StreamingOutputCallRequest.newBuilder()
      .addResponseParameters(ResponseParameters.newBuilder().setSize(1))
      .build();

  MethodDescriptor<StreamingOutputCallRequest, StreamingOutputCallResponse> md =
      TestServiceGrpc.getStreamingOutputCallMethod();
  ByteSizeMarshaller<StreamingOutputCallResponse> mar =
      new ByteSizeMarshaller<StreamingOutputCallResponse>(md.getResponseMarshaller());
  blockingServerStreamingCall(
      blockingStub.getChannel(),
      md.toBuilder(md.getRequestMarshaller(), mar).build(),
      blockingStub.getCallOptions(),
      request)
      .next();

  int size = mar.lastInSize;

  TestServiceGrpc.TestServiceBlockingStub stub =
      blockingStub.withMaxInboundMessageSize(size);

  stub.streamingOutputCall(request).next();
}

@Test
public void maxOutboundSize_exact() {
  StreamingOutputCallRequest request = StreamingOutputCallRequest.newBuilder()
      .addResponseParameters(ResponseParameters.newBuilder().setSize(1))
      .build();

  MethodDescriptor<StreamingOutputCallRequest, StreamingOutputCallResponse> md =
      TestServiceGrpc.getStreamingOutputCallMethod();
  ByteSizeMarshaller<StreamingOutputCallRequest> mar =
      new ByteSizeMarshaller<StreamingOutputCallRequest>(md.getRequestMarshaller());
  blockingServerStreamingCall(
      blockingStub.getChannel(),
      md.toBuilder(mar, md.getResponseMarshaller()).build(),
      blockingStub.getCallOptions(),
      request)
      .next();

  int size = mar.lastOutSize;

  TestServiceGrpc.TestServiceBlockingStub stub =
      blockingStub.withMaxOutboundMessageSize(size);

  stub.streamingOutputCall(request).next();
}

NettyClientStream(
    TransportState state,
    MethodDescriptor<?, ?> method,
    Metadata headers,
    Channel channel,
    AsciiString authority,
    AsciiString scheme,
    AsciiString userAgent,
    StatsTraceContext statsTraceCtx,
    TransportTracer transportTracer) {
  super(
      new NettyWritableBufferAllocator(channel.alloc()),
      statsTraceCtx,
      transportTracer,
      headers,
      useGet(method));
  this.state = checkNotNull(state, "transportState");
  this.writeQueue = state.handler.getWriteQueue();
  this.method = checkNotNull(method, "method");
  this.channel = checkNotNull(channel, "channel");
  this.authority = checkNotNull(authority, "authority");
  this.scheme = checkNotNull(scheme, "scheme");
  this.userAgent = userAgent;
}

/**
 * Generate a JWT-specific service URI. The URI is simply an identifier with enough information
 * for a service to know that the JWT was intended for it. The URI will commonly be verified with
 * a simple string equality check.
 */
private URI serviceUri(Channel channel, MethodDescriptor<?, ?> method) throws StatusException {
  String authority = channel.authority();
  if (authority == null) {
    throw Status.UNAUTHENTICATED.withDescription("Channel has no authority").asException();
  }
  // Always use HTTPS, by definition.
  final String scheme = "https";
  final int defaultPort = 443;
  String path = "/" + MethodDescriptor.extractFullServiceName(method.getFullMethodName());
  URI uri;
  try {
    uri = new URI(scheme, authority, path, null, null);
  } catch (URISyntaxException e) {
    throw Status.UNAUTHENTICATED.withDescription("Unable to construct service URI for auth")
        .withCause(e).asException();
  }
  // The default port must not be present. Alternative ports should be present.
  if (uri.getPort() == defaultPort) {
    uri = removePort(uri);
  }
  return uri;
}

/**
 * Generate a JWT-specific service URI. The URI is simply an identifier with enough information
 * for a service to know that the JWT was intended for it. The URI will commonly be verified with
 * a simple string equality check.
 */
private static URI serviceUri(String authority, MethodDescriptor<?, ?> method)
    throws StatusException {
  if (authority == null) {
    throw Status.UNAUTHENTICATED.withDescription("Channel has no authority").asException();
  }
  // Always use HTTPS, by definition.
  final String scheme = "https";
  final int defaultPort = 443;
  String path = "/" + MethodDescriptor.extractFullServiceName(method.getFullMethodName());
  URI uri;
  try {
    uri = new URI(scheme, authority, path, null, null);
  } catch (URISyntaxException e) {
    throw Status.UNAUTHENTICATED.withDescription("Unable to construct service URI for auth")
        .withCause(e).asException();
  }
  // The default port must not be present. Alternative ports should be present.
  if (uri.getPort() == defaultPort) {
    uri = removePort(uri);
  }
  return uri;
}

@Test
public void serviceMethodDescriotrs() {
  MethodDescriptor<SimpleRequest, SimpleResponse> genericTypeShouldMatchWhenAssigned;

  genericTypeShouldMatchWhenAssigned = SimpleServiceGrpc.getUnaryRpcMethod();
  assertEquals(UNARY, genericTypeShouldMatchWhenAssigned.getType());

  genericTypeShouldMatchWhenAssigned = SimpleServiceGrpc.getClientStreamingRpcMethod();
  assertEquals(CLIENT_STREAMING, genericTypeShouldMatchWhenAssigned.getType());

  genericTypeShouldMatchWhenAssigned = SimpleServiceGrpc.getServerStreamingRpcMethod();
  assertEquals(SERVER_STREAMING, genericTypeShouldMatchWhenAssigned.getType());

  genericTypeShouldMatchWhenAssigned = SimpleServiceGrpc.getBidiStreamingRpcMethod();
  assertEquals(BIDI_STREAMING, genericTypeShouldMatchWhenAssigned.getType());
}

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

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

ClientCallImpl(
    MethodDescriptor<ReqT, RespT> method, Executor executor, CallOptions callOptions,
    ClientTransportProvider clientTransportProvider,
    ScheduledExecutorService deadlineCancellationExecutor,
    CallTracer channelCallsTracer) {
  this.method = method;
  // If we know that the executor is a direct executor, we don't need to wrap it with a
  // SerializingExecutor. This is purely for performance reasons.
  // See https://github.com/grpc/grpc-java/issues/368
  this.callExecutor = executor == directExecutor()
      ? new SerializeReentrantCallsDirectExecutor()
      : new SerializingExecutor(executor);
  this.channelCallsTracer = channelCallsTracer;
  // Propagate the context from the thread which initiated the call to all callbacks.
  this.context = Context.current();
  this.unaryRequest = method.getType() == MethodType.UNARY
      || method.getType() == MethodType.SERVER_STREAMING;
  this.callOptions = callOptions;
  this.clientTransportProvider = clientTransportProvider;
  this.deadlineCancellationExecutor = deadlineCancellationExecutor;
}

@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 deadlineExceededBeforeCallStarted() {
  CallOptions callOptions = baseCallOptions.withDeadlineAfter(0, TimeUnit.SECONDS);
  fakeClock.forwardTime(System.nanoTime(), TimeUnit.NANOSECONDS);
  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(transport, times(0))
      .newStream(any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class));
  verify(callListener, timeout(1000)).onClose(statusCaptor.capture(), any(Metadata.class));
  assertEquals(Status.Code.DEADLINE_EXCEEDED, statusCaptor.getValue().getCode());
  verifyZeroInteractions(provider);
}

private void sendMessage_serverSendsOne_closeOnSecondCall(
    MethodDescriptor<Long, Long> method) {
  ServerCallImpl<Long, Long> serverCall = new ServerCallImpl<Long, Long>(
      stream,
      method,
      requestHeaders,
      context,
      DecompressorRegistry.getDefaultInstance(),
      CompressorRegistry.getDefaultInstance());
  serverCall.sendHeaders(new Metadata());
  serverCall.sendMessage(1L);
  verify(stream, times(1)).writeMessage(any(InputStream.class));
  verify(stream, never()).close(any(Status.class), any(Metadata.class));

  // trying to send a second message causes gRPC to close the underlying stream
  serverCall.sendMessage(1L);
  verify(stream, times(1)).writeMessage(any(InputStream.class));
  ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
  ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
  verify(stream, times(1)).close(statusCaptor.capture(), metadataCaptor.capture());
  assertEquals(Status.Code.INTERNAL, statusCaptor.getValue().getCode());
  assertEquals(ServerCallImpl.TOO_MANY_RESPONSES, statusCaptor.getValue().getDescription());
  assertTrue(metadataCaptor.getValue().keys().isEmpty());
}

private void sendMessage_serverSendsOne_closeOnSecondCall_appRunToCompletion(
    MethodDescriptor<Long, Long> method) {
  ServerCallImpl<Long, Long> serverCall = new ServerCallImpl<Long, Long>(
      stream,
      method,
      requestHeaders,
      context,
      DecompressorRegistry.getDefaultInstance(),
      CompressorRegistry.getDefaultInstance());
  serverCall.sendHeaders(new Metadata());
  serverCall.sendMessage(1L);
  serverCall.sendMessage(1L);
  verify(stream, times(1)).writeMessage(any(InputStream.class));
  verify(stream, times(1)).close(any(Status.class), any(Metadata.class));

  // App runs to completion but everything is ignored
  serverCall.sendMessage(1L);
  serverCall.close(Status.OK, new Metadata());
  try {
    serverCall.close(Status.OK, new Metadata());
    fail("calling a second time should still cause an error");
  } catch (IllegalStateException expected) {
    // noop
  }
}

private void serverSendsOne_okFailsOnMissingResponse(
    MethodDescriptor<Long, Long> method) {
  ServerCallImpl<Long, Long> serverCall = new ServerCallImpl<Long, Long>(
      stream,
      method,
      requestHeaders,
      context,
      DecompressorRegistry.getDefaultInstance(),
      CompressorRegistry.getDefaultInstance());
  serverCall.close(Status.OK, new Metadata());
  ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
  ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
  verify(stream, times(1)).close(statusCaptor.capture(), metadataCaptor.capture());
  assertEquals(Status.Code.INTERNAL, statusCaptor.getValue().getCode());
  assertEquals(ServerCallImpl.MISSING_RESPONSE, statusCaptor.getValue().getDescription());
  assertTrue(metadataCaptor.getValue().keys().isEmpty());
}

@Test
public void reprocess_NoPendingStream() {
  SubchannelPicker picker = mock(SubchannelPicker.class);
  AbstractSubchannel subchannel = mock(AbstractSubchannel.class);
  when(subchannel.obtainActiveTransport()).thenReturn(mockRealTransport);
  when(picker.pickSubchannel(any(PickSubchannelArgs.class))).thenReturn(
      PickResult.withSubchannel(subchannel));
  when(mockRealTransport.newStream(any(MethodDescriptor.class), any(Metadata.class),
          any(CallOptions.class))).thenReturn(mockRealStream);
  delayedTransport.reprocess(picker);
  verifyNoMoreInteractions(picker);
  verifyNoMoreInteractions(transportListener);

  // Though picker was not originally used, it will be saved and serve future streams.
  ClientStream stream = delayedTransport.newStream(method, headers, CallOptions.DEFAULT);
  verify(picker).pickSubchannel(new PickSubchannelArgsImpl(method, headers, CallOptions.DEFAULT));
  verify(subchannel).obtainActiveTransport();
  assertSame(mockRealStream, stream);
}

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

@Before
public void setUp() throws Exception {
  MockitoAnnotations.initMocks(this);
  MethodDescriptor<String, Integer> flowMethod = MethodDescriptor.<String, Integer>newBuilder()
      .setType(MethodType.UNKNOWN)
      .setFullMethodName("basic/flow")
      .setRequestMarshaller(requestMarshaller)
      .setResponseMarshaller(responseMarshaller)
      .build();
  basicServiceDefinition = ServerServiceDefinition.builder(
      new ServiceDescriptor("basic", flowMethod))
      .addMethod(flowMethod, flowHandler)
      .build();

  MethodDescriptor<String, Integer> coupleMethod =
      flowMethod.toBuilder().setFullMethodName("multi/couple").build();
  MethodDescriptor<String, Integer> fewMethod =
      flowMethod.toBuilder().setFullMethodName("multi/few").build();
  multiServiceDefinition = ServerServiceDefinition.builder(
      new ServiceDescriptor("multi", coupleMethod, fewMethod))
      .addMethod(coupleMethod, coupleHandler)
      .addMethod(fewMethod, fewHandler)
      .build();

  flowMethodDefinition = getOnlyElement(basicServiceDefinition.getMethods());
}

@Test
public void replaceAndLookup() {
  assertNull(registry.addService(basicServiceDefinition));
  assertNotNull(registry.lookupMethod("basic/flow"));
  MethodDescriptor<String, Integer> anotherMethod = MethodDescriptor.<String, Integer>newBuilder()
      .setType(MethodType.UNKNOWN)
      .setFullMethodName("basic/another")
      .setRequestMarshaller(requestMarshaller)
      .setResponseMarshaller(responseMarshaller)
      .build();
  ServerServiceDefinition replaceServiceDefinition = ServerServiceDefinition.builder(
      new ServiceDescriptor("basic", anotherMethod))
      .addMethod(anotherMethod, flowHandler).build();
  ServerMethodDefinition<?, ?> anotherMethodDefinition =
      replaceServiceDefinition.getMethod("basic/another");
  assertSame(basicServiceDefinition, registry.addService(replaceServiceDefinition));

  assertNull(registry.lookupMethod("basic/flow"));
  ServerMethodDefinition<?, ?> method = registry.lookupMethod("basic/another");
  assertSame(anotherMethodDefinition, method);
}