AbstractStream(
    FirestoreChannel channel,
    MethodDescriptor<ReqT, RespT> methodDescriptor,
    AsyncQueue workerQueue,
    TimerId connectionTimerId,
    TimerId idleTimerId,
    CallbackT listener) {
  this.firestoreChannel = channel;
  this.methodDescriptor = methodDescriptor;
  this.workerQueue = workerQueue;
  this.idleTimerId = idleTimerId;
  this.listener = listener;
  this.idleTimeoutRunnable = new IdleTimeoutRunnable();

  backoff =
      new ExponentialBackoff(
          workerQueue,
          connectionTimerId,
          BACKOFF_INITIAL_DELAY_MS,
          BACKOFF_FACTOR,
          BACKOFF_MAX_DELAY_MS);
}
 

@Test
public void blockingUnaryCall_HasBlockingStubType() {
  NoopClientCall<Integer, Integer> call = new NoopClientCall<Integer, Integer>() {
    @Override
    public void start(io.grpc.ClientCall.Listener<Integer> listener, Metadata headers) {
      listener.onMessage(1);
      listener.onClose(Status.OK, new Metadata());
    }
  };
  when(mockChannel.newCall(
      ArgumentMatchers.<MethodDescriptor<Integer, Integer>>any(), any(CallOptions.class)))
      .thenReturn(call);

  Integer unused =
      ClientCalls.blockingUnaryCall(mockChannel, UNARY_METHOD, CallOptions.DEFAULT, 1);

  verify(mockChannel).newCall(methodDescriptorCaptor.capture(), callOptionsCaptor.capture());
  CallOptions capturedCallOption = callOptionsCaptor.getValue();
  assertThat(capturedCallOption.getOption(ClientCalls.STUB_TYPE_OPTION))
      .isEquivalentAccordingToCompareTo(StubType.BLOCKING);
}
 

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

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

@CheckForNull
private MethodInfo getMethodInfo(MethodDescriptor<?, ?> method) {
  ManagedChannelServiceConfig mcsc = managedChannelServiceConfig.get();
  if (mcsc == null) {
    return null;
  }
  MethodInfo info;
  info = mcsc.getServiceMethodMap().get(method.getFullMethodName());
  if (info == null) {
    String serviceName = method.getServiceName();
    info = mcsc.getServiceMap().get(serviceName);
  }
  if (info == null) {
    info = mcsc.getDefaultMethodConfig();
  }
  return info;
}
 

@Before
public void setUp() throws Exception {
    this.descriptor = MethodDescriptor.<String, Integer>newBuilder()
            .setType(MethodDescriptor.MethodType.CLIENT_STREAMING)
            .setFullMethodName("a.service/method")
            .setRequestMarshaller(stringMarshaller)
            .setResponseMarshaller(intMarshaller)
            .build();
    this.callOptions = CallOptions.DEFAULT;
    this.clientCall = new ClientCallRecorder();
    when(channel.newCall(descriptor, callOptions)).thenReturn(clientCall);

    discardEventListener = spy(new LoggingDiscardEventListener<String>(DiscardClientInterceptorTest.class.getName(), 1));
    this.interceptor = new DiscardClientInterceptor(discardEventListener, 1);

    this.call = (DiscardClientInterceptor.DiscardClientCall<String, Integer>) interceptor.interceptCall(descriptor, callOptions, channel);
}
 

<ReqT, RespT> StreamObserver<ReqT> callBidiStreamingWithRetry(
    BackOffer backOffer,
    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<>(backOffer);
  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;
}
 

/**
 * {@inheritDoc}
 */
@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
    final MethodDescriptor<ReqT, RespT> method,
    final CallOptions callOptions,
    final Channel next
) {
    final String methodType = method.getType().toString();
    final String methodName = method.getFullMethodName();
    final int channelId = next.hashCode();
    log.info("gRPC {} call: {} (channel: {})", methodType, methodName, channelId);
    return next.newCall(method, callOptions);
}
 

@Test
public void idempotentMethod_usesHttpPut() {
  SetStreamFactoryRunnable callback = new SetStreamFactoryRunnable(factory);
  MethodDescriptor<?, ?> idempotentMethod = method.toBuilder().setIdempotent(true).build();
  CronetClientStream stream =
      new CronetClientStream(
          "https://www.google.com:443",
          "cronet",
          executor,
          metadata,
          transport,
          callback,
          lock,
          100,
          false /* alwaysUsePut */,
          idempotentMethod,
          StatsTraceContext.NOOP,
          CallOptions.DEFAULT,
          transportTracer);
  callback.setStream(stream);
  ExperimentalBidirectionalStream.Builder builder =
      mock(ExperimentalBidirectionalStream.Builder.class);
  when(factory.newBidirectionalStreamBuilder(
          any(String.class), any(BidirectionalStream.Callback.class), any(Executor.class)))
      .thenReturn(builder);
  when(builder.build()).thenReturn(cronetStream);
  stream.start(clientListener);

  verify(builder).setHttpMethod("PUT");
}
 

/** Set up for test. */
@Before
public void startUp() {
  descriptor = MethodDescriptor.<String, Integer>newBuilder()
      .setType(MethodDescriptor.MethodType.UNKNOWN)
      .setFullMethodName("a.service/method")
      .setRequestMarshaller(stringMarshaller)
      .setResponseMarshaller(intMarshaller)
      .build();
  when(channel.newCall(same(descriptor), any(CallOptions.class))).thenReturn(call);
  doReturn("localhost:443").when(channel).authority();
  interceptor = new ClientAuthInterceptor(credentials, executor);
}
 

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
    MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
  return new CheckedForwardingClientCall<ReqT, RespT>(
      next.newCall(method.toBuilder().setSafe(true).build(), callOptions)) {
    @Override
    public void checkedStart(Listener<RespT> responseListener, Metadata headers) {
      delegate().start(responseListener, headers);
    }
  };
}
 

@Test
public void pickerReturnsStreamTracer_delayed() {
  ClientStream mockStream = mock(ClientStream.class);
  ClientStreamTracer.Factory factory1 = mock(ClientStreamTracer.Factory.class);
  ClientStreamTracer.Factory factory2 = mock(ClientStreamTracer.Factory.class);
  createChannel();

  CallOptions callOptions = CallOptions.DEFAULT.withStreamTracerFactory(factory1);
  ClientCall<String, Integer> call = channel.newCall(method, callOptions);
  call.start(mockCallListener, new Metadata());

  Subchannel subchannel =
      createSubchannelSafely(helper, addressGroup, Attributes.EMPTY, subchannelStateListener);
  requestConnectionSafely(helper, subchannel);
  MockClientTransportInfo transportInfo = transports.poll();
  transportInfo.listener.transportReady();
  ClientTransport mockTransport = transportInfo.transport;
  when(mockTransport.newStream(
          any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class)))
      .thenReturn(mockStream);
  when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class))).thenReturn(
      PickResult.withSubchannel(subchannel, factory2));

  updateBalancingStateSafely(helper, READY, mockPicker);
  assertEquals(1, executor.runDueTasks());

  verify(mockPicker).pickSubchannel(any(PickSubchannelArgs.class));
  verify(mockTransport).newStream(same(method), any(Metadata.class), callOptionsCaptor.capture());
  assertEquals(
      Arrays.asList(factory1, factory2),
      callOptionsCaptor.getValue().getStreamTracerFactories());
  // The factories are safely not stubbed because we do not expect any usage of them.
  verifyZeroInteractions(factory1);
  verifyZeroInteractions(factory2);
}
 

/**
 * Sets up mocks.
 */
@Before public void setUp() {
  MockitoAnnotations.initMocks(this);
  ClientCall<SimpleRequest, SimpleResponse> call =
      new NoopClientCall<>();
  when(channel.newCall(
          ArgumentMatchers.<MethodDescriptor<SimpleRequest, SimpleResponse>>any(),
          any(CallOptions.class)))
      .thenReturn(call);
}
 

/**
 * 
 * @param method
 * @param respStream
 * @param responseExecutor
 */
public ResilientBiDiStream(MethodDescriptor<ReqT,RespT> method,
        ResilientResponseObserver<ReqT,RespT> respStream,
        Executor responseExecutor) {
    this.method = method;
    this.respStream = respStream;
    this.responseExecutor = serialized(responseExecutor != null
            ? responseExecutor : userExecutor);
    this.requestExecutor = sendViaEventLoop ? serialized(ses) : null;
}
 

/**
 * Extracts the method name from the given method.
 *
 * @param method The method to get the method name from.
 * @return The extracted method name.
 * @see #extractServiceName(MethodDescriptor)
 */
public static String extractMethodName(final MethodDescriptor<?, ?> method) {
    // This method is the equivalent of MethodDescriptor.extractFullServiceName
    final String fullMethodName = method.getFullMethodName();
    final int index = fullMethodName.lastIndexOf('/');
    if (index == -1) {
        return fullMethodName;
    }
    return fullMethodName.substring(index + 1);
}
 

/**
 * If a {@link SubchannelPicker} is being, or has been provided via {@link #reprocess}, the last
 * picker will be consulted.
 *
 * <p>Otherwise, if the delayed transport is not shutdown, then a {@link PendingStream} is
 * returned; if the transport is shutdown, then a {@link FailingClientStream} is returned.
 */
@Override
public final ClientStream newStream(
    MethodDescriptor<?, ?> method, Metadata headers, CallOptions callOptions) {
  try {
    PickSubchannelArgs args = new PickSubchannelArgsImpl(method, headers, callOptions);
    SubchannelPicker picker = null;
    long pickerVersion = -1;
    while (true) {
      synchronized (lock) {
        if (shutdownStatus != null) {
          return new FailingClientStream(shutdownStatus);
        }
        if (lastPicker == null) {
          return createPendingStream(args);
        }
        // Check for second time through the loop, and whether anything changed
        if (picker != null && pickerVersion == lastPickerVersion) {
          return createPendingStream(args);
        }
        picker = lastPicker;
        pickerVersion = lastPickerVersion;
      }
      PickResult pickResult = picker.pickSubchannel(args);
      ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult,
          callOptions.isWaitForReady());
      if (transport != null) {
        return transport.newStream(
            args.getMethodDescriptor(), args.getHeaders(), args.getCallOptions());
      }
      // This picker's conclusion is "buffer".  If there hasn't been a newer picker set (possible
      // race with reprocess()), we will buffer it.  Otherwise, will try with the new picker.
    }
  } finally {
    syncContext.drain();
  }
}
 

@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) {
            String token = getToken(next);
            if (token != null) {
                headers.put(TOKEN, token);
            }
            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);
        }
    };
}
 

@Test
public void getRequestSentThroughHeader() {
  // Creating a GET method
  MethodDescriptor<?, ?> descriptor = MethodDescriptor.<Void, Void>newBuilder()
      .setType(MethodDescriptor.MethodType.UNARY)
      .setFullMethodName("testService/test")
      .setRequestMarshaller(marshaller)
      .setResponseMarshaller(marshaller)
      .setIdempotent(true)
      .setSafe(true)
      .build();
  NettyClientStream stream = new NettyClientStream(
      new TransportStateImpl(handler, DEFAULT_MAX_MESSAGE_SIZE),
      descriptor,
      new Metadata(),
      channel,
      AsciiString.of("localhost"),
      AsciiString.of("http"),
      AsciiString.of("agent"),
      StatsTraceContext.NOOP,
      transportTracer,
      CallOptions.DEFAULT);
  stream.start(listener);
  stream.transportState().setId(STREAM_ID);
  stream.transportState().setHttp2Stream(http2Stream);

  byte[] msg = smallMessage();
  stream.writeMessage(new ByteArrayInputStream(msg));
  stream.flush();
  stream.halfClose();
  ArgumentCaptor<CreateStreamCommand> cmdCap = ArgumentCaptor.forClass(CreateStreamCommand.class);
  verify(writeQueue).enqueue(cmdCap.capture(), eq(true));
  ImmutableListMultimap<CharSequence, CharSequence> headers =
      ImmutableListMultimap.copyOf(cmdCap.getValue().headers());
  assertThat(headers).containsEntry(AsciiString.of(":method"), Utils.HTTP_GET_METHOD);
  assertThat(headers)
      .containsEntry(
          AsciiString.of(":path"),
          AsciiString.of("/testService/test?" + BaseEncoding.base64().encode(msg)));
}
 

@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 testTracedServerWithCustomOperationName() {
  TracingServerInterceptor tracingInterceptor =
      TracingServerInterceptor.newBuilder()
          .withTracer(serverTracer)
          .withOperationName(
              new OperationNameConstructor() {
                @Override
                public <ReqT, RespT> String constructOperationName(
                    MethodDescriptor<ReqT, RespT> method) {
                  return PREFIX + method.getFullMethodName();
                }
              })
          .build();
  TracedService.addGeeterService(grpcServer.getServiceRegistry(), tracingInterceptor);

  assertEquals("call should complete successfully", "Hello world", client.greet().getMessage());
  await().atMost(5, TimeUnit.SECONDS).until(reportedSpansSize(serverTracer), equalTo(1));
  assertEquals(
      "one span should have been created and finished for one client request",
      serverTracer.finishedSpans().size(),
      1);

  MockSpan span = serverTracer.finishedSpans().get(0);
  assertEquals(
      "span should have prefix", span.operationName(), PREFIX + "helloworld.Greeter/SayHello");
  assertEquals("span should have no parents", span.parentId(), 0);
  assertEquals("span should have no logs", span.logEntries().size(), 0);
  Assertions.assertThat(span.tags())
      .as("span should have base server tags")
      .isEqualTo(BASE_SERVER_TAGS);
  assertFalse("span should have no baggage", span.context().baggageItems().iterator().hasNext());
}
 

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,
        CallOptions callOptions,
        Channel next) {
    log.info(method.getFullMethodName());
    return next.newCall(method, callOptions);
}
 

@SuppressWarnings({"rawtypes", "unchecked"})
private LoggingInterceptor getInterceptorWithAlwaysThisHandler(
    LoggingHandler handler, AsynchronousFileOutputStream outputFile) {
  return new LoggingInterceptor(outputFile, clock) {
    @Override
    public <ReqT, RespT> LoggingHandler<ReqT, RespT> selectHandler(
        MethodDescriptor<ReqT, RespT> method) {
      return handler;
    }
  };
}
 

@Test
public void maxOutboundSize_tooBig() {
  // set at least one field to ensure the size is non-zero.
  StreamingOutputCallRequest request = StreamingOutputCallRequest.newBuilder()
      .addResponseParameters(ResponseParameters.newBuilder().setSize(1))
      .build();


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

  TestServiceGrpc.TestServiceBlockingStub stub =
      blockingStub.withMaxOutboundMessageSize(mar.lastOutSize - 1);
  try {
    stub.streamingOutputCall(request).next();
    fail();
  } catch (StatusRuntimeException ex) {
    Status s = ex.getStatus();
    assertWithMessage(s.toString()).that(s.getCode()).isEqualTo(Status.Code.CANCELLED);
    assertThat(Throwables.getStackTraceAsString(ex)).contains("message too large");
  }
}
 

/**
 * Gets all method names from the given service descriptor.
 *
 * @param serviceDescriptor The service descriptor to get the names from.
 * @return The newly created and sorted list of the method names.
 */
protected List<String> collectMethodNamesForService(final ServiceDescriptor serviceDescriptor) {
    final List<String> methods = new ArrayList<>();
    for (final MethodDescriptor<?, ?> grpcMethod : serviceDescriptor.getMethods()) {
        methods.add(extractMethodName(grpcMethod));
    }
    methods.sort(String.CASE_INSENSITIVE_ORDER);
    return methods;
}
 

@Override
@Nullable
public Span startSpan(@Nullable AbstractSpan<?> parent,
                      @Nullable MethodDescriptor<?, ?> method,
                      @Nullable String authority) {

    if (null == parent) {
        return null;
    }

    // we only support unary method calls and ignore others for now
    if (method != null && method.getType() != MethodDescriptor.MethodType.UNARY) {
        return null;
    }

    Span span = parent.createExitSpan();
    if (span == null) {
        // as it's an external call, we only need a single span for nested calls
        return null;
    }

    span.withName(method == null ? null : method.getFullMethodName())
        .withType("external")
        .withSubtype(GRPC);

    if (authority != null) {
        Destination destination = span.getContext().getDestination()
            .withAddressPort(authority);

        destination.getService()
            .withName(GRPC)
            .withResource(authority)
            .withType(GRPC);
    }
    return span.activate();
}
 

ServerCallImpl(ServerStream stream, MethodDescriptor<ReqT, RespT> method,
    Metadata inboundHeaders, Context.CancellableContext context,
    DecompressorRegistry decompressorRegistry, CompressorRegistry compressorRegistry,
    CallTracer serverCallTracer, Tag tag) {
  this.stream = stream;
  this.method = method;
  this.context = context;
  this.messageAcceptEncoding = inboundHeaders.get(MESSAGE_ACCEPT_ENCODING_KEY);
  this.decompressorRegistry = decompressorRegistry;
  this.compressorRegistry = compressorRegistry;
  this.serverCallTracer = serverCallTracer;
  this.serverCallTracer.reportCallStarted();
  this.tag = tag;
}
 

private ServiceDescriptor getServiceDescriptor(ServerServiceDefinition template, ProviderConfig providerConfig,
                                               List<MethodDescriptor<Request, Response>> methodDescriptors) {
    String serviceName = providerConfig.getInterfaceId();
    ServiceDescriptor.Builder builder = ServiceDescriptor.newBuilder(serviceName)
        .setSchemaDescriptor(template.getServiceDescriptor().getSchemaDescriptor());
    for (MethodDescriptor<Request, Response> methodDescriptor : methodDescriptors) {
        builder.addMethod(methodDescriptor);
    }
    return builder.build();

}
 

/**
 * 将服务注册与注销相关的参数封装成一个对象
 *
 * @author sxp
 * @since V1.0 2017/3/21
 */
private List<Map<String,Object>> createServerParams() {
  int port = transportServer.getPort();

  List<ServerServiceDefinition> serviceDfs = getServices();

  List<Map<String,Object>> params = new ArrayList<>(serviceDfs.size());
  StringBuilder sb = new StringBuilder();
  Map<String, Object> oneService;
  Collection<MethodDescriptor<?, ?>> methodDesps;
  String methodName;

  for (ServerServiceDefinition item : serviceDfs) {
    sb.setLength(0);

    methodDesps = item.getServiceDescriptor().getMethods();
    for (MethodDescriptor<?, ?> md : methodDesps) {
      methodName = GrpcUtils.getSimpleMethodName(md.getFullMethodName());
      sb.append(methodName);
      sb.append(",");// 多个方法之间用英文逗号分隔
    }
    sb.deleteCharAt(sb.lastIndexOf(","));

    oneService = new HashMap<>();
    oneService.put(GlobalConstants.Provider.Key.INTERFACE, item.getServiceDescriptor().getName());
    oneService.put(GlobalConstants.CommonKey.METHODS, sb.toString());
    oneService.put(GlobalConstants.PROVIDER_SERVICE_PORT, port);

    params.add(oneService);
  }

  return params;
}
 

@Override
protected Counter newRequestCounterFor(final MethodDescriptor<?, ?> method) {
    return this.counterCustomizer.apply(
            prepareCounterFor(method,
                    METRIC_NAME_SERVER_REQUESTS_RECEIVED,
                    "The total number of requests received"))
            .register(this.registry);
}
 

@Override
public ClientStream newStream(
    MethodDescriptor<?, ?> method, Metadata headers, CallOptions callOptions) {
  Preconditions.checkNotNull(method, "method");
  Preconditions.checkNotNull(headers, "headers");
  if (channel == null) {
    return new FailingClientStream(statusExplainingWhyTheChannelIsNull);
  }
  StatsTraceContext statsTraceCtx =
      StatsTraceContext.newClientContext(callOptions, getAttributes(), headers);
  return new NettyClientStream(
      new NettyClientStream.TransportState(
          handler,
          channel.eventLoop(),
          maxMessageSize,
          statsTraceCtx,
          transportTracer,
          method.getFullMethodName()) {
        @Override
        protected Status statusFromFailedFuture(ChannelFuture f) {
          return NettyClientTransport.this.statusFromFailedFuture(f);
        }
      },
      method,
      headers,
      channel,
      authority,
      negotiationScheme,
      userAgent,
      statsTraceCtx,
      transportTracer,
      callOptions,
      useGetForSafeMethods);
}