core/src/main/java/io/grpc/internal/ClientCallImpl.java - platform/external/grpc-grpc-java - Gitiles

 /*
  * Copyright 2014, Google Inc. All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are
  * met:
  *
  *    * Redistributions of source code must retain the above copyright
  * notice, this list of conditions and the following disclaimer.
  *    * Redistributions in binary form must reproduce the above
  * copyright notice, this list of conditions and the following disclaimer
  * in the documentation and/or other materials provided with the
  * distribution.
  *
  *    * Neither the name of Google Inc. nor the names of its
  * contributors may be used to endorse or promote products derived from
  * this software without specific prior written permission.
  *
  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  */

 package io.grpc.internal;

 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 import static io.grpc.Contexts.statusFromCancelled;
 import static io.grpc.Status.DEADLINE_EXCEEDED;
 import static io.grpc.internal.GrpcUtil.MESSAGE_ACCEPT_ENCODING_KEY;
 import static io.grpc.internal.GrpcUtil.MESSAGE_ENCODING_KEY;
 import static io.grpc.internal.GrpcUtil.TIMEOUT_KEY;
 import static java.lang.Math.max;

 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;

 import io.grpc.Attributes;
 import io.grpc.CallOptions;
 import io.grpc.ClientCall;
 import io.grpc.Codec;
 import io.grpc.Compressor;
 import io.grpc.CompressorRegistry;
 import io.grpc.Context;
 import io.grpc.Deadline;
 import io.grpc.Decompressor;
 import io.grpc.DecompressorRegistry;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.MethodDescriptor.MethodType;
 import io.grpc.Status;

 import java.io.InputStream;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;

 import javax.annotation.Nullable;

 /**
  * Implementation of {@link ClientCall}.
  */
 final class ClientCallImpl<ReqT, RespT> extends ClientCall<ReqT, RespT>
     implements Context.CancellationListener {

   private static final Logger log = Logger.getLogger(ClientCallImpl.class.getName());

   private final MethodDescriptor<ReqT, RespT> method;
   private final Executor callExecutor;
   private final Context context;
   private volatile ScheduledFuture<?> deadlineCancellationFuture;
   private final boolean unaryRequest;
   private final CallOptions callOptions;
   private final StatsTraceContext statsTraceCtx;
   private ClientStream stream;
   private volatile boolean cancelListenersShouldBeRemoved;
   private boolean cancelCalled;
   private boolean halfCloseCalled;
   private final ClientTransportProvider clientTransportProvider;
   private ScheduledExecutorService deadlineCancellationExecutor;
   private DecompressorRegistry decompressorRegistry = DecompressorRegistry.getDefaultInstance();
   private CompressorRegistry compressorRegistry = CompressorRegistry.getDefaultInstance();

   ClientCallImpl(MethodDescriptor<ReqT, RespT> method, Executor executor,
       CallOptions callOptions, StatsTraceContext statsTraceCtx,
       ClientTransportProvider clientTransportProvider,
       ScheduledExecutorService deadlineCancellationExecutor) {
     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);
     // Propagate the context from the thread which initiated the call to all callbacks.
     this.context = Context.current();
     this.statsTraceCtx = Preconditions.checkNotNull(statsTraceCtx, "statsTraceCtx");
     this.unaryRequest = method.getType() == MethodType.UNARY
         || method.getType() == MethodType.SERVER_STREAMING;
     this.callOptions = callOptions;
     this.clientTransportProvider = clientTransportProvider;
     this.deadlineCancellationExecutor = deadlineCancellationExecutor;
   }

   @Override
   public void cancelled(Context context) {
     stream.cancel(statusFromCancelled(context));
   }

   /**
    * Provider of {@link ClientTransport}s.
    */
   interface ClientTransportProvider {
     /**
      * Returns a transport for a new call.
      */
     ClientTransport get(CallOptions callOptions, Metadata headers);
   }

   ClientCallImpl<ReqT, RespT> setDecompressorRegistry(DecompressorRegistry decompressorRegistry) {
     this.decompressorRegistry = decompressorRegistry;
     return this;
   }

   ClientCallImpl<ReqT, RespT> setCompressorRegistry(CompressorRegistry compressorRegistry) {
     this.compressorRegistry = compressorRegistry;
     return this;
   }

   @VisibleForTesting
   static void prepareHeaders(Metadata headers, DecompressorRegistry decompressorRegistry,
       Compressor compressor, StatsTraceContext statsTraceCtx) {
     headers.discardAll(MESSAGE_ENCODING_KEY);
     if (compressor != Codec.Identity.NONE) {
       headers.put(MESSAGE_ENCODING_KEY, compressor.getMessageEncoding());
     }

     headers.discardAll(MESSAGE_ACCEPT_ENCODING_KEY);
     String advertisedEncodings = decompressorRegistry.getRawAdvertisedMessageEncodings();
     if (!advertisedEncodings.isEmpty()) {
       headers.put(MESSAGE_ACCEPT_ENCODING_KEY, advertisedEncodings);
     }
     statsTraceCtx.propagateToHeaders(headers);
   }

   @Override
   public void start(final Listener<RespT> observer, Metadata headers) {
     checkState(stream == null, "Already started");
     checkNotNull(observer, "observer");
     checkNotNull(headers, "headers");

     if (context.isCancelled()) {
       // Context is already cancelled so no need to create a real stream, just notify the observer
       // of cancellation via callback on the executor
       stream = NoopClientStream.INSTANCE;
       class ClosedByContext extends ContextRunnable {
         ClosedByContext() {
           super(context);
         }

         @Override
         public void runInContext() {
           closeObserver(observer, statusFromCancelled(context), new Metadata());
         }
       }

       callExecutor.execute(new ClosedByContext());
       return;
     }
     final String compressorName = callOptions.getCompressor();
     Compressor compressor = null;
     if (compressorName != null) {
       compressor = compressorRegistry.lookupCompressor(compressorName);
       if (compressor == null) {
         stream = NoopClientStream.INSTANCE;
         class ClosedByNotFoundCompressor extends ContextRunnable {
           ClosedByNotFoundCompressor() {
             super(context);
           }

           @Override
           public void runInContext() {
             closeObserver(
                 observer,
                 Status.INTERNAL.withDescription(
                     String.format("Unable to find compressor by name %s", compressorName)),
                 new Metadata());
           }
         }

         callExecutor.execute(new ClosedByNotFoundCompressor());
         return;
       }
     } else {
       compressor = Codec.Identity.NONE;
     }

     prepareHeaders(headers, decompressorRegistry, compressor, statsTraceCtx);

     Deadline effectiveDeadline = effectiveDeadline();
     boolean deadlineExceeded = effectiveDeadline != null && effectiveDeadline.isExpired();
     if (!deadlineExceeded) {
       updateTimeoutHeaders(effectiveDeadline, callOptions.getDeadline(),
           context.getDeadline(), headers);
       ClientTransport transport = clientTransportProvider.get(callOptions, headers);
       Context origContext = context.attach();
       try {
         stream = transport.newStream(method, headers, callOptions, statsTraceCtx);
       } finally {
         context.detach(origContext);
       }
     } else {
       stream = new FailingClientStream(DEADLINE_EXCEEDED);
     }

     if (callOptions.getAuthority() != null) {
       stream.setAuthority(callOptions.getAuthority());
     }
     if (callOptions.getMaxInboundMessageSize() != null) {
       stream.setMaxInboundMessageSize(callOptions.getMaxInboundMessageSize());
     }
     if (callOptions.getMaxOutboundMessageSize() != null) {
       stream.setMaxOutboundMessageSize(callOptions.getMaxOutboundMessageSize());
     }
     stream.setCompressor(compressor);
     stream.start(new ClientStreamListenerImpl(observer));

     // Delay any sources of cancellation after start(), because most of the transports are broken if
     // they receive cancel before start. Issue #1343 has more details

     // Propagate later Context cancellation to the remote side.
     context.addListener(this, directExecutor());
     if (effectiveDeadline != null
         // If the context has the effective deadline, we don't need to schedule an extra task.
         && context.getDeadline() != effectiveDeadline) {
       deadlineCancellationFuture = startDeadlineTimer(effectiveDeadline);
     }
     if (cancelListenersShouldBeRemoved) {
       // Race detected! ClientStreamListener.closed may have been called before
       // deadlineCancellationFuture was set / context listener added, thereby preventing the future
       // and listener from being cancelled. Go ahead and cancel again, just to be sure it
       // was cancelled.
       removeContextListenerAndCancelDeadlineFuture();
     }
   }

   /**
    * Based on the deadline, calculate and set the timeout to the given headers.
    */
   private static void updateTimeoutHeaders(@Nullable Deadline effectiveDeadline,
       @Nullable Deadline callDeadline, @Nullable Deadline outerCallDeadline, Metadata headers) {
     headers.discardAll(TIMEOUT_KEY);

     if (effectiveDeadline == null) {
       return;
     }

     long effectiveTimeout = max(0, effectiveDeadline.timeRemaining(TimeUnit.NANOSECONDS));
     headers.put(TIMEOUT_KEY, effectiveTimeout);

     logIfContextNarrowedTimeout(effectiveTimeout, effectiveDeadline, outerCallDeadline,
         callDeadline);
   }

   private static void logIfContextNarrowedTimeout(long effectiveTimeout,
       Deadline effectiveDeadline, @Nullable Deadline outerCallDeadline,
       @Nullable Deadline callDeadline) {
     if (!log.isLoggable(Level.INFO) || outerCallDeadline != effectiveDeadline) {
       return;
     }

     StringBuilder builder = new StringBuilder();
     builder.append(String.format("Call timeout set to '%d' ns, due to context deadline.",
         effectiveTimeout));
     if (callDeadline == null) {
       builder.append(" Explicit call timeout was not set.");
     } else {
       long callTimeout = callDeadline.timeRemaining(TimeUnit.NANOSECONDS);
       builder.append(String.format(" Explicit call timeout was '%d' ns.", callTimeout));
     }

     log.info(builder.toString());
   }

   private void removeContextListenerAndCancelDeadlineFuture() {
     context.removeListener(this);
     ScheduledFuture<?> f = deadlineCancellationFuture;
     if (f != null) {
       f.cancel(false);
     }
   }

   private class DeadlineTimer implements Runnable {
     @Override
     public void run() {
       // DelayedStream.cancel() is safe to call from a thread that is different from where the
       // stream is created.
       stream.cancel(DEADLINE_EXCEEDED);
     }
   }

   private ScheduledFuture<?> startDeadlineTimer(Deadline deadline) {
     return deadlineCancellationExecutor.schedule(
         new LogExceptionRunnable(new DeadlineTimer()), deadline.timeRemaining(TimeUnit.NANOSECONDS),
         TimeUnit.NANOSECONDS);
   }

   @Nullable
   private Deadline effectiveDeadline() {
     // Call options and context are immutable, so we don't need to cache the deadline.
     return min(callOptions.getDeadline(), context.getDeadline());
   }

   @Nullable
   private static Deadline min(@Nullable Deadline deadline0, @Nullable Deadline deadline1) {
     if (deadline0 == null) {
       return deadline1;
     }
     if (deadline1 == null) {
       return deadline0;
     }
     return deadline0.minimum(deadline1);
   }

   @Override
   public void request(int numMessages) {
     Preconditions.checkState(stream != null, "Not started");
     checkArgument(numMessages >= 0, "Number requested must be non-negative");
     stream.request(numMessages);
   }

   @Override
   public void cancel(@Nullable String message, @Nullable Throwable cause) {
     if (message == null && cause == null) {
       cause = new CancellationException("Cancelled without a message or cause");
       log.log(Level.WARNING, "Cancelling without a message or cause is suboptimal", cause);
     }
     if (cancelCalled) {
       return;
     }
     cancelCalled = true;
     try {
       // Cancel is called in exception handling cases, so it may be the case that the
       // stream was never successfully created or start has never been called.
       if (stream != null) {
         Status status = Status.CANCELLED;
         if (message != null) {
           status = status.withDescription(message);
         }
         if (cause != null) {
           status = status.withCause(cause);
         }
         stream.cancel(status);
       }
     } finally {
       removeContextListenerAndCancelDeadlineFuture();
     }
   }

   @Override
   public void halfClose() {
     Preconditions.checkState(stream != null, "Not started");
     Preconditions.checkState(!cancelCalled, "call was cancelled");
     Preconditions.checkState(!halfCloseCalled, "call already half-closed");
     halfCloseCalled = true;
     stream.halfClose();
   }

   @Override
   public void sendMessage(ReqT message) {
     Preconditions.checkState(stream != null, "Not started");
     Preconditions.checkState(!cancelCalled, "call was cancelled");
     Preconditions.checkState(!halfCloseCalled, "call was half-closed");
     try {
       // TODO(notcarl): Find out if messageIs needs to be closed.
       InputStream messageIs = method.streamRequest(message);
       stream.writeMessage(messageIs);
     } catch (Throwable e) {
       stream.cancel(Status.CANCELLED.withCause(e).withDescription("Failed to stream message"));
       return;
     }
     // For unary requests, we don't flush since we know that halfClose should be coming soon. This
     // allows us to piggy-back the END_STREAM=true on the last message frame without opening the
     // possibility of broken applications forgetting to call halfClose without noticing.
     if (!unaryRequest) {
       stream.flush();
     }
   }

   @Override
   public void setMessageCompression(boolean enabled) {
     checkState(stream != null, "Not started");
     stream.setMessageCompression(enabled);
   }

   @Override
   public boolean isReady() {
     return stream.isReady();
   }

   @Override
   public Attributes getAttributes() {
     if (stream != null) {
       return stream.getAttributes();
     }
     return Attributes.EMPTY;
   }

   private void closeObserver(Listener<RespT> observer, Status status, Metadata trailers) {
     statsTraceCtx.callEnded(status);
     observer.onClose(status, trailers);
   }

   private class ClientStreamListenerImpl implements ClientStreamListener {
     private final Listener<RespT> observer;
     private boolean closed;

     public ClientStreamListenerImpl(Listener<RespT> observer) {
       this.observer = Preconditions.checkNotNull(observer, "observer");
     }

     @Override
     public void headersRead(final Metadata headers) {
       Decompressor decompressor = Codec.Identity.NONE;
       if (headers.containsKey(MESSAGE_ENCODING_KEY)) {
         String encoding = headers.get(MESSAGE_ENCODING_KEY);
         decompressor = decompressorRegistry.lookupDecompressor(encoding);
         if (decompressor == null) {
           stream.cancel(Status.INTERNAL.withDescription(
               String.format("Can't find decompressor for %s", encoding)));
           return;
         }
       }
       stream.setDecompressor(decompressor);

       class HeadersRead extends ContextRunnable {
         HeadersRead() {
           super(context);
         }

         @Override
         public final void runInContext() {
           try {
             if (closed) {
               return;
             }
             observer.onHeaders(headers);
           } catch (Throwable t) {
             Status status =
                 Status.CANCELLED.withCause(t).withDescription("Failed to read headers");
             stream.cancel(status);
             close(status, new Metadata());
           }
         }
       }

       callExecutor.execute(new HeadersRead());
     }

     @Override
     public void messageRead(final InputStream message) {
       class MessageRead extends ContextRunnable {
         MessageRead() {
           super(context);
         }

         @Override
         public final void runInContext() {
           try {
             if (closed) {
               return;
             }
             try {
               observer.onMessage(method.parseResponse(message));
             } finally {
               message.close();
             }
           } catch (Throwable t) {
             Status status =
                 Status.CANCELLED.withCause(t).withDescription("Failed to read message.");
             stream.cancel(status);
             close(status, new Metadata());
           }
         }
       }

       callExecutor.execute(new MessageRead());
     }

     /**
      * Must be called from application thread.
      */
     private void close(Status status, Metadata trailers) {
       closed = true;
       cancelListenersShouldBeRemoved = true;
       try {
         closeObserver(observer, status, trailers);
       } finally {
         removeContextListenerAndCancelDeadlineFuture();
       }
     }

     @Override
     public void closed(Status status, Metadata trailers) {
       Deadline deadline = effectiveDeadline();
       if (status.getCode() == Status.Code.CANCELLED && deadline != null) {
         // When the server's deadline expires, it can only reset the stream with CANCEL and no
         // description. Since our timer may be delayed in firing, we double-check the deadline and
         // turn the failure into the likely more helpful DEADLINE_EXCEEDED status.
         if (deadline.isExpired()) {
           status = DEADLINE_EXCEEDED;
           // Replace trailers to prevent mixing sources of status and trailers.
           trailers = new Metadata();
         }
       }
       final Status savedStatus = status;
       final Metadata savedTrailers = trailers;
       class StreamClosed extends ContextRunnable {
         StreamClosed() {
           super(context);
         }

         @Override
         public final void runInContext() {
           if (closed) {
             // We intentionally don't keep the status or metadata from the server.
             return;
           }
           close(savedStatus, savedTrailers);
         }
       }

       callExecutor.execute(new StreamClosed());
     }

     @Override
     public void onReady() {
       class StreamOnReady extends ContextRunnable {
         StreamOnReady() {
           super(context);
         }

         @Override
         public final void runInContext() {
           try {
             observer.onReady();
           } catch (Throwable t) {
             Status status =
                 Status.CANCELLED.withCause(t).withDescription("Failed to call onReady.");
             stream.cancel(status);
             close(status, new Metadata());
           }
         }
       }

       callExecutor.execute(new StreamOnReady());
     }
   }
 }
	/*
	* Copyright 2014, Google Inc. All rights reserved.
	*
	* Redistribution and use in source and binary forms, with or without
	* modification, are permitted provided that the following conditions are
	* met:
	*
	* * Redistributions of source code must retain the above copyright
	* notice, this list of conditions and the following disclaimer.
	* * Redistributions in binary form must reproduce the above
	* copyright notice, this list of conditions and the following disclaimer
	* in the documentation and/or other materials provided with the
	* distribution.
	*
	* * Neither the name of Google Inc. nor the names of its
	* contributors may be used to endorse or promote products derived from
	* this software without specific prior written permission.
	*
	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
	* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
	* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
	* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
	* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
	* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
	* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
	* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
	* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
	* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
	* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
	*/

	package io.grpc.internal;

	import static com.google.common.base.Preconditions.checkArgument;
	import static com.google.common.base.Preconditions.checkNotNull;
	import static com.google.common.base.Preconditions.checkState;
	import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
	import static io.grpc.Contexts.statusFromCancelled;
	import static io.grpc.Status.DEADLINE_EXCEEDED;
	import static io.grpc.internal.GrpcUtil.MESSAGE_ACCEPT_ENCODING_KEY;
	import static io.grpc.internal.GrpcUtil.MESSAGE_ENCODING_KEY;
	import static io.grpc.internal.GrpcUtil.TIMEOUT_KEY;
	import static java.lang.Math.max;

	import com.google.common.annotations.VisibleForTesting;
	import com.google.common.base.Preconditions;

	import io.grpc.Attributes;
	import io.grpc.CallOptions;
	import io.grpc.ClientCall;
	import io.grpc.Codec;
	import io.grpc.Compressor;
	import io.grpc.CompressorRegistry;
	import io.grpc.Context;
	import io.grpc.Deadline;
	import io.grpc.Decompressor;
	import io.grpc.DecompressorRegistry;
	import io.grpc.Metadata;
	import io.grpc.MethodDescriptor;
	import io.grpc.MethodDescriptor.MethodType;
	import io.grpc.Status;

	import java.io.InputStream;
	import java.util.concurrent.CancellationException;
	import java.util.concurrent.Executor;
	import java.util.concurrent.ScheduledExecutorService;
	import java.util.concurrent.ScheduledFuture;
	import java.util.concurrent.TimeUnit;
	import java.util.logging.Level;
	import java.util.logging.Logger;

	import javax.annotation.Nullable;

	/**
	* Implementation of {@link ClientCall}.
	*/
	final class ClientCallImpl<ReqT, RespT> extends ClientCall<ReqT, RespT>
	implements Context.CancellationListener {

	private static final Logger log = Logger.getLogger(ClientCallImpl.class.getName());

	private final MethodDescriptor<ReqT, RespT> method;
	private final Executor callExecutor;
	private final Context context;
	private volatile ScheduledFuture<?> deadlineCancellationFuture;
	private final boolean unaryRequest;
	private final CallOptions callOptions;
	private final StatsTraceContext statsTraceCtx;
	private ClientStream stream;
	private volatile boolean cancelListenersShouldBeRemoved;
	private boolean cancelCalled;
	private boolean halfCloseCalled;
	private final ClientTransportProvider clientTransportProvider;
	private ScheduledExecutorService deadlineCancellationExecutor;
	private DecompressorRegistry decompressorRegistry = DecompressorRegistry.getDefaultInstance();
	private CompressorRegistry compressorRegistry = CompressorRegistry.getDefaultInstance();

	ClientCallImpl(MethodDescriptor<ReqT, RespT> method, Executor executor,
	CallOptions callOptions, StatsTraceContext statsTraceCtx,
	ClientTransportProvider clientTransportProvider,
	ScheduledExecutorService deadlineCancellationExecutor) {
	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);
	// Propagate the context from the thread which initiated the call to all callbacks.
	this.context = Context.current();
	this.statsTraceCtx = Preconditions.checkNotNull(statsTraceCtx, "statsTraceCtx");
	this.unaryRequest = method.getType() == MethodType.UNARY
	\|\| method.getType() == MethodType.SERVER_STREAMING;
	this.callOptions = callOptions;
	this.clientTransportProvider = clientTransportProvider;
	this.deadlineCancellationExecutor = deadlineCancellationExecutor;
	}

	@Override
	public void cancelled(Context context) {
	stream.cancel(statusFromCancelled(context));
	}

	/**
	* Provider of {@link ClientTransport}s.
	*/
	interface ClientTransportProvider {
	/**
	* Returns a transport for a new call.
	*/
	ClientTransport get(CallOptions callOptions, Metadata headers);
	}

	ClientCallImpl<ReqT, RespT> setDecompressorRegistry(DecompressorRegistry decompressorRegistry) {
	this.decompressorRegistry = decompressorRegistry;
	return this;
	}

	ClientCallImpl<ReqT, RespT> setCompressorRegistry(CompressorRegistry compressorRegistry) {
	this.compressorRegistry = compressorRegistry;
	return this;
	}

	@VisibleForTesting
	static void prepareHeaders(Metadata headers, DecompressorRegistry decompressorRegistry,
	Compressor compressor, StatsTraceContext statsTraceCtx) {
	headers.discardAll(MESSAGE_ENCODING_KEY);
	if (compressor != Codec.Identity.NONE) {
	headers.put(MESSAGE_ENCODING_KEY, compressor.getMessageEncoding());
	}

	headers.discardAll(MESSAGE_ACCEPT_ENCODING_KEY);
	String advertisedEncodings = decompressorRegistry.getRawAdvertisedMessageEncodings();
	if (!advertisedEncodings.isEmpty()) {
	headers.put(MESSAGE_ACCEPT_ENCODING_KEY, advertisedEncodings);
	}
	statsTraceCtx.propagateToHeaders(headers);
	}

	@Override
	public void start(final Listener<RespT> observer, Metadata headers) {
	checkState(stream == null, "Already started");
	checkNotNull(observer, "observer");
	checkNotNull(headers, "headers");

	if (context.isCancelled()) {
	// Context is already cancelled so no need to create a real stream, just notify the observer
	// of cancellation via callback on the executor
	stream = NoopClientStream.INSTANCE;
	class ClosedByContext extends ContextRunnable {
	ClosedByContext() {
	super(context);
	}

	@Override
	public void runInContext() {
	closeObserver(observer, statusFromCancelled(context), new Metadata());
	}
	}

	callExecutor.execute(new ClosedByContext());
	return;
	}
	final String compressorName = callOptions.getCompressor();
	Compressor compressor = null;
	if (compressorName != null) {
	compressor = compressorRegistry.lookupCompressor(compressorName);
	if (compressor == null) {
	stream = NoopClientStream.INSTANCE;
	class ClosedByNotFoundCompressor extends ContextRunnable {
	ClosedByNotFoundCompressor() {
	super(context);
	}

	@Override
	public void runInContext() {
	closeObserver(
	observer,
	Status.INTERNAL.withDescription(
	String.format("Unable to find compressor by name %s", compressorName)),
	new Metadata());
	}
	}

	callExecutor.execute(new ClosedByNotFoundCompressor());
	return;
	}
	} else {
	compressor = Codec.Identity.NONE;
	}

	prepareHeaders(headers, decompressorRegistry, compressor, statsTraceCtx);

	Deadline effectiveDeadline = effectiveDeadline();
	boolean deadlineExceeded = effectiveDeadline != null && effectiveDeadline.isExpired();
	if (!deadlineExceeded) {
	updateTimeoutHeaders(effectiveDeadline, callOptions.getDeadline(),
	context.getDeadline(), headers);
	ClientTransport transport = clientTransportProvider.get(callOptions, headers);
	Context origContext = context.attach();
	try {
	stream = transport.newStream(method, headers, callOptions, statsTraceCtx);
	} finally {
	context.detach(origContext);
	}
	} else {
	stream = new FailingClientStream(DEADLINE_EXCEEDED);
	}

	if (callOptions.getAuthority() != null) {
	stream.setAuthority(callOptions.getAuthority());
	}
	if (callOptions.getMaxInboundMessageSize() != null) {
	stream.setMaxInboundMessageSize(callOptions.getMaxInboundMessageSize());
	}
	if (callOptions.getMaxOutboundMessageSize() != null) {
	stream.setMaxOutboundMessageSize(callOptions.getMaxOutboundMessageSize());
	}
	stream.setCompressor(compressor);
	stream.start(new ClientStreamListenerImpl(observer));

	// Delay any sources of cancellation after start(), because most of the transports are broken if
	// they receive cancel before start. Issue #1343 has more details

	// Propagate later Context cancellation to the remote side.
	context.addListener(this, directExecutor());
	if (effectiveDeadline != null
	// If the context has the effective deadline, we don't need to schedule an extra task.
	&& context.getDeadline() != effectiveDeadline) {
	deadlineCancellationFuture = startDeadlineTimer(effectiveDeadline);
	}
	if (cancelListenersShouldBeRemoved) {
	// Race detected! ClientStreamListener.closed may have been called before
	// deadlineCancellationFuture was set / context listener added, thereby preventing the future
	// and listener from being cancelled. Go ahead and cancel again, just to be sure it
	// was cancelled.
	removeContextListenerAndCancelDeadlineFuture();
	}
	}

	/**
	* Based on the deadline, calculate and set the timeout to the given headers.
	*/
	private static void updateTimeoutHeaders(@Nullable Deadline effectiveDeadline,
	@Nullable Deadline callDeadline, @Nullable Deadline outerCallDeadline, Metadata headers) {
	headers.discardAll(TIMEOUT_KEY);

	if (effectiveDeadline == null) {
	return;
	}

	long effectiveTimeout = max(0, effectiveDeadline.timeRemaining(TimeUnit.NANOSECONDS));
	headers.put(TIMEOUT_KEY, effectiveTimeout);

	logIfContextNarrowedTimeout(effectiveTimeout, effectiveDeadline, outerCallDeadline,
	callDeadline);
	}

	private static void logIfContextNarrowedTimeout(long effectiveTimeout,
	Deadline effectiveDeadline, @Nullable Deadline outerCallDeadline,
	@Nullable Deadline callDeadline) {
	if (!log.isLoggable(Level.INFO) \|\| outerCallDeadline != effectiveDeadline) {
	return;
	}

	StringBuilder builder = new StringBuilder();
	builder.append(String.format("Call timeout set to '%d' ns, due to context deadline.",
	effectiveTimeout));
	if (callDeadline == null) {
	builder.append(" Explicit call timeout was not set.");
	} else {
	long callTimeout = callDeadline.timeRemaining(TimeUnit.NANOSECONDS);
	builder.append(String.format(" Explicit call timeout was '%d' ns.", callTimeout));
	}

	log.info(builder.toString());
	}

	private void removeContextListenerAndCancelDeadlineFuture() {
	context.removeListener(this);
	ScheduledFuture<?> f = deadlineCancellationFuture;
	if (f != null) {
	f.cancel(false);
	}
	}

	private class DeadlineTimer implements Runnable {
	@Override
	public void run() {
	// DelayedStream.cancel() is safe to call from a thread that is different from where the
	// stream is created.
	stream.cancel(DEADLINE_EXCEEDED);
	}
	}

	private ScheduledFuture<?> startDeadlineTimer(Deadline deadline) {
	return deadlineCancellationExecutor.schedule(
	new LogExceptionRunnable(new DeadlineTimer()), deadline.timeRemaining(TimeUnit.NANOSECONDS),
	TimeUnit.NANOSECONDS);
	}

	@Nullable
	private Deadline effectiveDeadline() {
	// Call options and context are immutable, so we don't need to cache the deadline.
	return min(callOptions.getDeadline(), context.getDeadline());
	}

	@Nullable
	private static Deadline min(@Nullable Deadline deadline0, @Nullable Deadline deadline1) {
	if (deadline0 == null) {
	return deadline1;
	}
	if (deadline1 == null) {
	return deadline0;
	}
	return deadline0.minimum(deadline1);
	}

	@Override
	public void request(int numMessages) {
	Preconditions.checkState(stream != null, "Not started");
	checkArgument(numMessages >= 0, "Number requested must be non-negative");
	stream.request(numMessages);
	}

	@Override
	public void cancel(@Nullable String message, @Nullable Throwable cause) {
	if (message == null && cause == null) {
	cause = new CancellationException("Cancelled without a message or cause");
	log.log(Level.WARNING, "Cancelling without a message or cause is suboptimal", cause);
	}
	if (cancelCalled) {
	return;
	}
	cancelCalled = true;
	try {
	// Cancel is called in exception handling cases, so it may be the case that the
	// stream was never successfully created or start has never been called.
	if (stream != null) {
	Status status = Status.CANCELLED;
	if (message != null) {
	status = status.withDescription(message);
	}
	if (cause != null) {
	status = status.withCause(cause);
	}
	stream.cancel(status);
	}
	} finally {
	removeContextListenerAndCancelDeadlineFuture();
	}
	}

	@Override
	public void halfClose() {
	Preconditions.checkState(stream != null, "Not started");
	Preconditions.checkState(!cancelCalled, "call was cancelled");
	Preconditions.checkState(!halfCloseCalled, "call already half-closed");
	halfCloseCalled = true;
	stream.halfClose();
	}

	@Override
	public void sendMessage(ReqT message) {
	Preconditions.checkState(stream != null, "Not started");
	Preconditions.checkState(!cancelCalled, "call was cancelled");
	Preconditions.checkState(!halfCloseCalled, "call was half-closed");
	try {
	// TODO(notcarl): Find out if messageIs needs to be closed.
	InputStream messageIs = method.streamRequest(message);
	stream.writeMessage(messageIs);
	} catch (Throwable e) {
	stream.cancel(Status.CANCELLED.withCause(e).withDescription("Failed to stream message"));
	return;
	}
	// For unary requests, we don't flush since we know that halfClose should be coming soon. This
	// allows us to piggy-back the END_STREAM=true on the last message frame without opening the
	// possibility of broken applications forgetting to call halfClose without noticing.
	if (!unaryRequest) {
	stream.flush();
	}
	}

	@Override
	public void setMessageCompression(boolean enabled) {
	checkState(stream != null, "Not started");
	stream.setMessageCompression(enabled);
	}

	@Override
	public boolean isReady() {
	return stream.isReady();
	}

	@Override
	public Attributes getAttributes() {
	if (stream != null) {
	return stream.getAttributes();
	}
	return Attributes.EMPTY;
	}

	private void closeObserver(Listener<RespT> observer, Status status, Metadata trailers) {
	statsTraceCtx.callEnded(status);
	observer.onClose(status, trailers);
	}

	private class ClientStreamListenerImpl implements ClientStreamListener {
	private final Listener<RespT> observer;
	private boolean closed;

	public ClientStreamListenerImpl(Listener<RespT> observer) {
	this.observer = Preconditions.checkNotNull(observer, "observer");
	}

	@Override
	public void headersRead(final Metadata headers) {
	Decompressor decompressor = Codec.Identity.NONE;
	if (headers.containsKey(MESSAGE_ENCODING_KEY)) {
	String encoding = headers.get(MESSAGE_ENCODING_KEY);
	decompressor = decompressorRegistry.lookupDecompressor(encoding);
	if (decompressor == null) {
	stream.cancel(Status.INTERNAL.withDescription(
	String.format("Can't find decompressor for %s", encoding)));
	return;
	}
	}
	stream.setDecompressor(decompressor);

	class HeadersRead extends ContextRunnable {
	HeadersRead() {
	super(context);
	}

	@Override
	public final void runInContext() {
	try {
	if (closed) {
	return;
	}
	observer.onHeaders(headers);
	} catch (Throwable t) {
	Status status =
	Status.CANCELLED.withCause(t).withDescription("Failed to read headers");
	stream.cancel(status);
	close(status, new Metadata());
	}
	}
	}

	callExecutor.execute(new HeadersRead());
	}

	@Override
	public void messageRead(final InputStream message) {
	class MessageRead extends ContextRunnable {
	MessageRead() {
	super(context);
	}

	@Override
	public final void runInContext() {
	try {
	if (closed) {
	return;
	}
	try {
	observer.onMessage(method.parseResponse(message));
	} finally {
	message.close();
	}
	} catch (Throwable t) {
	Status status =
	Status.CANCELLED.withCause(t).withDescription("Failed to read message.");
	stream.cancel(status);
	close(status, new Metadata());
	}
	}
	}

	callExecutor.execute(new MessageRead());
	}

	/**
	* Must be called from application thread.
	*/
	private void close(Status status, Metadata trailers) {
	closed = true;
	cancelListenersShouldBeRemoved = true;
	try {
	closeObserver(observer, status, trailers);
	} finally {
	removeContextListenerAndCancelDeadlineFuture();
	}
	}

	@Override
	public void closed(Status status, Metadata trailers) {
	Deadline deadline = effectiveDeadline();
	if (status.getCode() == Status.Code.CANCELLED && deadline != null) {
	// When the server's deadline expires, it can only reset the stream with CANCEL and no
	// description. Since our timer may be delayed in firing, we double-check the deadline and
	// turn the failure into the likely more helpful DEADLINE_EXCEEDED status.
	if (deadline.isExpired()) {
	status = DEADLINE_EXCEEDED;
	// Replace trailers to prevent mixing sources of status and trailers.
	trailers = new Metadata();
	}
	}
	final Status savedStatus = status;
	final Metadata savedTrailers = trailers;
	class StreamClosed extends ContextRunnable {
	StreamClosed() {
	super(context);
	}

	@Override
	public final void runInContext() {
	if (closed) {
	// We intentionally don't keep the status or metadata from the server.
	return;
	}
	close(savedStatus, savedTrailers);
	}
	}

	callExecutor.execute(new StreamClosed());
	}

	@Override
	public void onReady() {
	class StreamOnReady extends ContextRunnable {
	StreamOnReady() {
	super(context);
	}

	@Override
	public final void runInContext() {
	try {
	observer.onReady();
	} catch (Throwable t) {
	Status status =
	Status.CANCELLED.withCause(t).withDescription("Failed to call onReady.");
	stream.cancel(status);
	close(status, new Metadata());
	}
	}
	}

	callExecutor.execute(new StreamOnReady());
	}
	}
	}