ClientCallImpl extends ClientCall<ReqT, RespT> implements Context.CancellationListener { @Override public void start(final Listener<RespT> observer, Metadata headers) { checkState(stream == null, "Already started"); checkNotNull(observer, "observer"); checkNotNull(headers, "headers"); final Deadline effectiveDeadline = min(callOptions.getDeadline(), parentContext.getDeadline()); if (effectiveDeadline != parentContext.getDeadline()) { context = parentContext.withDeadline(effectiveDeadline, deadlineCancellationExecutor); } else { context = parentContext.withCancellation(); } if (context.isCancelled()) { stream = NoopClientStream.INSTANCE; callExecutor.execute(new ContextRunnable(context) { @Override public void runInContext() { observer.onClose(statusFromCancelled(context), new Metadata()); } }); return; } final String compressorName = callOptions.getCompressor(); Compressor compressor = null; if (compressorName != null) { compressor = compressorRegistry.lookupCompressor(compressorName); if (compressor == null) { stream = NoopClientStream.INSTANCE; callExecutor.execute(new ContextRunnable(context) { @Override public void runInContext() { observer.onClose( Status.INTERNAL.withDescription( String.format("Unable to find compressor by name %s", compressorName)), new Metadata()); } }); return; } } else { compressor = Codec.Identity.NONE; } prepareHeaders(headers, callOptions, userAgent, decompressorRegistry, compressor); final boolean deadlineExceeded = effectiveDeadline != null && effectiveDeadline.isExpired(); if (!deadlineExceeded) { updateTimeoutHeaders(effectiveDeadline, callOptions.getDeadline(), parentContext.getDeadline(), headers); ClientTransport transport = clientTransportProvider.get(callOptions); stream = transport.newStream(method, headers); } else { stream = new FailingClientStream(DEADLINE_EXCEEDED); } if (callOptions.getAuthority() != null) { stream.setAuthority(callOptions.getAuthority()); } stream.setCompressor(compressor); stream.start(new ClientStreamListenerImpl(observer)); if (compressor != Codec.Identity.NONE) { stream.setMessageCompression(true); } context.addListener(this, directExecutor()); if (contextListenerShouldBeRemoved) { context.removeListener(this); } } ClientCallImpl(MethodDescriptor<ReqT, RespT> method, Executor executor,
      CallOptions callOptions, ClientTransportProvider clientTransportProvider,
      ScheduledExecutorService deadlineCancellationExecutor); @Override void cancelled(Context context); @Override void start(final Listener<RespT> observer, Metadata headers); @Override void request(int numMessages); @Override void cancel(@Nullable String message, @Nullable Throwable cause); @Override void halfClose(); @Override void sendMessage(ReqT message); @Override void setMessageCompression(boolean enabled); @Override boolean isReady();  }
@Test public void contextDeadlineShouldBePropagatedInMetadata() { long deadlineNanos = TimeUnit.SECONDS.toNanos(1); Context context = Context.current().withDeadlineAfter(deadlineNanos, TimeUnit.NANOSECONDS, deadlineCancellationExecutor); context.attach(); ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>( DESCRIPTOR, MoreExecutors.directExecutor(), CallOptions.DEFAULT, provider, deadlineCancellationExecutor); Metadata headers = new Metadata(); call.start(callListener, headers); assertTrue(headers.containsKey(GrpcUtil.TIMEOUT_KEY)); Long timeout = headers.get(GrpcUtil.TIMEOUT_KEY); assertNotNull(timeout); long deltaNanos = TimeUnit.MILLISECONDS.toNanos(400); assertTimeoutBetween(timeout, deadlineNanos - deltaNanos, deadlineNanos); }
@Test public void contextDeadlineShouldOverrideLargerMetadataTimeout() { long deadlineNanos = TimeUnit.SECONDS.toNanos(1); Context context = Context.current().withDeadlineAfter(deadlineNanos, TimeUnit.NANOSECONDS, deadlineCancellationExecutor); context.attach(); CallOptions callOpts = CallOptions.DEFAULT.withDeadlineAfter(2, TimeUnit.SECONDS); ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>( DESCRIPTOR, MoreExecutors.directExecutor(), callOpts, provider, deadlineCancellationExecutor); Metadata headers = new Metadata(); call.start(callListener, headers); assertTrue(headers.containsKey(GrpcUtil.TIMEOUT_KEY)); Long timeout = headers.get(GrpcUtil.TIMEOUT_KEY); assertNotNull(timeout); long deltaNanos = TimeUnit.MILLISECONDS.toNanos(400); assertTimeoutBetween(timeout, deadlineNanos - deltaNanos, deadlineNanos); }
@Test public void contextDeadlineShouldNotOverrideSmallerMetadataTimeout() { long deadlineNanos = TimeUnit.SECONDS.toNanos(2); Context context = Context.current().withDeadlineAfter(deadlineNanos, TimeUnit.NANOSECONDS, deadlineCancellationExecutor); context.attach(); CallOptions callOpts = CallOptions.DEFAULT.withDeadlineAfter(1, TimeUnit.SECONDS); ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>( DESCRIPTOR, MoreExecutors.directExecutor(), callOpts, provider, deadlineCancellationExecutor); Metadata headers = new Metadata(); call.start(callListener, headers); assertTrue(headers.containsKey(GrpcUtil.TIMEOUT_KEY)); Long timeout = headers.get(GrpcUtil.TIMEOUT_KEY); assertNotNull(timeout); long callOptsNanos = TimeUnit.SECONDS.toNanos(1); long deltaNanos = TimeUnit.MILLISECONDS.toNanos(400); assertTimeoutBetween(timeout, callOptsNanos - deltaNanos, callOptsNanos); }
@Test public void timeoutShouldNotBeSet() { ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>( DESCRIPTOR, MoreExecutors.directExecutor(), CallOptions.DEFAULT, provider, deadlineCancellationExecutor); Metadata headers = new Metadata(); call.start(callListener, headers); assertFalse(headers.containsKey(GrpcUtil.TIMEOUT_KEY)); }