@Test public void contextDeadlineShouldBePropagatedToStream() { Context context = Context.current() .withDeadlineAfter(1000, TimeUnit.MILLISECONDS, deadlineCancellationExecutor); Context origContext = context.attach(); ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>( method, MoreExecutors.directExecutor(), baseCallOptions, provider, deadlineCancellationExecutor, channelCallTracer, false ); call.start(callListener, new Metadata()); context.detach(origContext); ArgumentCaptor<Deadline> deadlineCaptor = ArgumentCaptor.forClass(Deadline.class); verify(stream).setDeadline(deadlineCaptor.capture()); assertTimeoutBetween(deadlineCaptor.getValue().timeRemaining(TimeUnit.MILLISECONDS), 600, 1000); }
@Override public void start(final Listener<RespT> observer, Metadata headers) { checkState(stream == null, "Already started"); checkState(!cancelCalled, "call was cancelled"); checkNotNull(observer, "observer"); checkNotNull(headers, "headers"); if (context.isCancelled()) { 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, fullStreamDecompression); Deadline effectiveDeadline = effectiveDeadline(); boolean deadlineExceeded = effectiveDeadline != null && effectiveDeadline.isExpired(); if (!deadlineExceeded) { logIfContextNarrowedTimeout( effectiveDeadline, callOptions.getDeadline(), context.getDeadline()); if (retryEnabled) { stream = clientTransportProvider.newRetriableStream(method, callOptions, headers, context); } else { ClientTransport transport = clientTransportProvider.get( new PickSubchannelArgsImpl(method, headers, callOptions)); Context origContext = context.attach(); try { stream = transport.newStream(method, headers, callOptions); } finally { context.detach(origContext); } } } else { stream = new FailingClientStream( DEADLINE_EXCEEDED.withDescription("deadline exceeded: " + effectiveDeadline)); } if (callOptions.getAuthority() != null) { stream.setAuthority(callOptions.getAuthority()); } if (callOptions.getMaxInboundMessageSize() != null) { stream.setMaxInboundMessageSize(callOptions.getMaxInboundMessageSize()); } if (callOptions.getMaxOutboundMessageSize() != null) { stream.setMaxOutboundMessageSize(callOptions.getMaxOutboundMessageSize()); } if (effectiveDeadline != null) { stream.setDeadline(effectiveDeadline); } stream.setCompressor(compressor); if (fullStreamDecompression) { stream.setFullStreamDecompression(fullStreamDecompression); } stream.setDecompressorRegistry(decompressorRegistry); channelCallsTracer.reportCallStarted(); stream.start(new ClientStreamListenerImpl(observer)); context.addListener(cancellationListener, directExecutor()); if (effectiveDeadline != null && context.getDeadline() != effectiveDeadline && deadlineCancellationExecutor != null) { deadlineCancellationFuture = startDeadlineTimer(effectiveDeadline); } if (cancelListenersShouldBeRemoved) { removeContextListenerAndCancelDeadlineFuture(); } }
ClientCallImpl extends ClientCall<ReqT, RespT> { @Override public void start(final Listener<RespT> observer, Metadata headers) { checkState(stream == null, "Already started"); checkState(!cancelCalled, "call was cancelled"); checkNotNull(observer, "observer"); checkNotNull(headers, "headers"); if (context.isCancelled()) { 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, fullStreamDecompression); Deadline effectiveDeadline = effectiveDeadline(); boolean deadlineExceeded = effectiveDeadline != null && effectiveDeadline.isExpired(); if (!deadlineExceeded) { logIfContextNarrowedTimeout( effectiveDeadline, callOptions.getDeadline(), context.getDeadline()); if (retryEnabled) { stream = clientTransportProvider.newRetriableStream(method, callOptions, headers, context); } else { ClientTransport transport = clientTransportProvider.get( new PickSubchannelArgsImpl(method, headers, callOptions)); Context origContext = context.attach(); try { stream = transport.newStream(method, headers, callOptions); } finally { context.detach(origContext); } } } else { stream = new FailingClientStream( DEADLINE_EXCEEDED.withDescription("deadline exceeded: " + effectiveDeadline)); } if (callOptions.getAuthority() != null) { stream.setAuthority(callOptions.getAuthority()); } if (callOptions.getMaxInboundMessageSize() != null) { stream.setMaxInboundMessageSize(callOptions.getMaxInboundMessageSize()); } if (callOptions.getMaxOutboundMessageSize() != null) { stream.setMaxOutboundMessageSize(callOptions.getMaxOutboundMessageSize()); } if (effectiveDeadline != null) { stream.setDeadline(effectiveDeadline); } stream.setCompressor(compressor); if (fullStreamDecompression) { stream.setFullStreamDecompression(fullStreamDecompression); } stream.setDecompressorRegistry(decompressorRegistry); channelCallsTracer.reportCallStarted(); stream.start(new ClientStreamListenerImpl(observer)); context.addListener(cancellationListener, directExecutor()); if (effectiveDeadline != null && context.getDeadline() != effectiveDeadline && deadlineCancellationExecutor != null) { deadlineCancellationFuture = startDeadlineTimer(effectiveDeadline); } if (cancelListenersShouldBeRemoved) { removeContextListenerAndCancelDeadlineFuture(); } } }
ClientCallImpl extends ClientCall<ReqT, RespT> { @Override public void start(final Listener<RespT> observer, Metadata headers) { checkState(stream == null, "Already started"); checkState(!cancelCalled, "call was cancelled"); checkNotNull(observer, "observer"); checkNotNull(headers, "headers"); if (context.isCancelled()) { 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, fullStreamDecompression); Deadline effectiveDeadline = effectiveDeadline(); boolean deadlineExceeded = effectiveDeadline != null && effectiveDeadline.isExpired(); if (!deadlineExceeded) { logIfContextNarrowedTimeout( effectiveDeadline, callOptions.getDeadline(), context.getDeadline()); if (retryEnabled) { stream = clientTransportProvider.newRetriableStream(method, callOptions, headers, context); } else { ClientTransport transport = clientTransportProvider.get( new PickSubchannelArgsImpl(method, headers, callOptions)); Context origContext = context.attach(); try { stream = transport.newStream(method, headers, callOptions); } finally { context.detach(origContext); } } } else { stream = new FailingClientStream( DEADLINE_EXCEEDED.withDescription("deadline exceeded: " + effectiveDeadline)); } if (callOptions.getAuthority() != null) { stream.setAuthority(callOptions.getAuthority()); } if (callOptions.getMaxInboundMessageSize() != null) { stream.setMaxInboundMessageSize(callOptions.getMaxInboundMessageSize()); } if (callOptions.getMaxOutboundMessageSize() != null) { stream.setMaxOutboundMessageSize(callOptions.getMaxOutboundMessageSize()); } if (effectiveDeadline != null) { stream.setDeadline(effectiveDeadline); } stream.setCompressor(compressor); if (fullStreamDecompression) { stream.setFullStreamDecompression(fullStreamDecompression); } stream.setDecompressorRegistry(decompressorRegistry); channelCallsTracer.reportCallStarted(); stream.start(new ClientStreamListenerImpl(observer)); context.addListener(cancellationListener, directExecutor()); if (effectiveDeadline != null && context.getDeadline() != effectiveDeadline && deadlineCancellationExecutor != null) { deadlineCancellationFuture = startDeadlineTimer(effectiveDeadline); } if (cancelListenersShouldBeRemoved) { removeContextListenerAndCancelDeadlineFuture(); } } ClientCallImpl(
      MethodDescriptor<ReqT, RespT> method, Executor executor, CallOptions callOptions,
      ClientTransportProvider clientTransportProvider,
      ScheduledExecutorService deadlineCancellationExecutor,
      CallTracer channelCallsTracer,
      boolean retryEnabled); }
ClientCallImpl extends ClientCall<ReqT, RespT> { @Override public void start(final Listener<RespT> observer, Metadata headers) { checkState(stream == null, "Already started"); checkState(!cancelCalled, "call was cancelled"); checkNotNull(observer, "observer"); checkNotNull(headers, "headers"); if (context.isCancelled()) { 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, fullStreamDecompression); Deadline effectiveDeadline = effectiveDeadline(); boolean deadlineExceeded = effectiveDeadline != null && effectiveDeadline.isExpired(); if (!deadlineExceeded) { logIfContextNarrowedTimeout( effectiveDeadline, callOptions.getDeadline(), context.getDeadline()); if (retryEnabled) { stream = clientTransportProvider.newRetriableStream(method, callOptions, headers, context); } else { ClientTransport transport = clientTransportProvider.get( new PickSubchannelArgsImpl(method, headers, callOptions)); Context origContext = context.attach(); try { stream = transport.newStream(method, headers, callOptions); } finally { context.detach(origContext); } } } else { stream = new FailingClientStream( DEADLINE_EXCEEDED.withDescription("deadline exceeded: " + effectiveDeadline)); } if (callOptions.getAuthority() != null) { stream.setAuthority(callOptions.getAuthority()); } if (callOptions.getMaxInboundMessageSize() != null) { stream.setMaxInboundMessageSize(callOptions.getMaxInboundMessageSize()); } if (callOptions.getMaxOutboundMessageSize() != null) { stream.setMaxOutboundMessageSize(callOptions.getMaxOutboundMessageSize()); } if (effectiveDeadline != null) { stream.setDeadline(effectiveDeadline); } stream.setCompressor(compressor); if (fullStreamDecompression) { stream.setFullStreamDecompression(fullStreamDecompression); } stream.setDecompressorRegistry(decompressorRegistry); channelCallsTracer.reportCallStarted(); stream.start(new ClientStreamListenerImpl(observer)); context.addListener(cancellationListener, directExecutor()); if (effectiveDeadline != null && context.getDeadline() != effectiveDeadline && deadlineCancellationExecutor != null) { deadlineCancellationFuture = startDeadlineTimer(effectiveDeadline); } if (cancelListenersShouldBeRemoved) { removeContextListenerAndCancelDeadlineFuture(); } } ClientCallImpl(
      MethodDescriptor<ReqT, RespT> method, Executor executor, CallOptions callOptions,
      ClientTransportProvider clientTransportProvider,
      ScheduledExecutorService deadlineCancellationExecutor,
      CallTracer channelCallsTracer,
      boolean retryEnabled); @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(); @Override Attributes getAttributes(); @Override String toString(); @Override String getFullMethod(); }
ClientCallImpl extends ClientCall<ReqT, RespT> { @Override public void start(final Listener<RespT> observer, Metadata headers) { checkState(stream == null, "Already started"); checkState(!cancelCalled, "call was cancelled"); checkNotNull(observer, "observer"); checkNotNull(headers, "headers"); if (context.isCancelled()) { 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, fullStreamDecompression); Deadline effectiveDeadline = effectiveDeadline(); boolean deadlineExceeded = effectiveDeadline != null && effectiveDeadline.isExpired(); if (!deadlineExceeded) { logIfContextNarrowedTimeout( effectiveDeadline, callOptions.getDeadline(), context.getDeadline()); if (retryEnabled) { stream = clientTransportProvider.newRetriableStream(method, callOptions, headers, context); } else { ClientTransport transport = clientTransportProvider.get( new PickSubchannelArgsImpl(method, headers, callOptions)); Context origContext = context.attach(); try { stream = transport.newStream(method, headers, callOptions); } finally { context.detach(origContext); } } } else { stream = new FailingClientStream( DEADLINE_EXCEEDED.withDescription("deadline exceeded: " + effectiveDeadline)); } if (callOptions.getAuthority() != null) { stream.setAuthority(callOptions.getAuthority()); } if (callOptions.getMaxInboundMessageSize() != null) { stream.setMaxInboundMessageSize(callOptions.getMaxInboundMessageSize()); } if (callOptions.getMaxOutboundMessageSize() != null) { stream.setMaxOutboundMessageSize(callOptions.getMaxOutboundMessageSize()); } if (effectiveDeadline != null) { stream.setDeadline(effectiveDeadline); } stream.setCompressor(compressor); if (fullStreamDecompression) { stream.setFullStreamDecompression(fullStreamDecompression); } stream.setDecompressorRegistry(decompressorRegistry); channelCallsTracer.reportCallStarted(); stream.start(new ClientStreamListenerImpl(observer)); context.addListener(cancellationListener, directExecutor()); if (effectiveDeadline != null && context.getDeadline() != effectiveDeadline && deadlineCancellationExecutor != null) { deadlineCancellationFuture = startDeadlineTimer(effectiveDeadline); } if (cancelListenersShouldBeRemoved) { removeContextListenerAndCancelDeadlineFuture(); } } ClientCallImpl(
      MethodDescriptor<ReqT, RespT> method, Executor executor, CallOptions callOptions,
      ClientTransportProvider clientTransportProvider,
      ScheduledExecutorService deadlineCancellationExecutor,
      CallTracer channelCallsTracer,
      boolean retryEnabled); @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(); @Override Attributes getAttributes(); @Override String toString(); @Override String getFullMethod();  }