Java 类io.grpc.Deadline 实例源码

/**
 * Prepares for a check that the subject is deadline within the given tolerance of an
 * expected value that will be provided in the next call in the fluent chain.
 */
@CheckReturnValue
public TolerantDeadlineComparison isWithin(final long delta, final TimeUnit timeUnit) {
  return new TolerantDeadlineComparison() {
    @Override
    public void of(Deadline expected) {
      Deadline actual = actual();
      checkNotNull(actual, "actual value cannot be null. expected=%s", expected);

      // This is probably overkill, but easier than thinking about overflow.
      BigInteger actualTimeRemaining = BigInteger.valueOf(actual.timeRemaining(NANOSECONDS));
      BigInteger expectedTimeRemaining = BigInteger.valueOf(expected.timeRemaining(NANOSECONDS));
      BigInteger deltaNanos = BigInteger.valueOf(timeUnit.toNanos(delta));
      if (actualTimeRemaining.subtract(expectedTimeRemaining).abs().compareTo(deltaNanos) > 0) {
        failWithRawMessage(
            "%s and <%s> should have been within <%sns> of each other",
            actualAsString(),
            expected,
            deltaNanos);
      }
    }
  };
}

private static void logIfContextNarrowedTimeout(long effectiveTimeout,
    Deadline effectiveDeadline, @Nullable Deadline outerCallDeadline,
    @Nullable Deadline callDeadline) {
  if (!log.isLoggable(Level.FINE) || 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.fine(builder.toString());
}

@Test
public void expiredDeadlineCancelsStream_CallOptions() {
  fakeClock.forwardTime(System.nanoTime(), TimeUnit.NANOSECONDS);
  // The deadline needs to be a number large enough to get encompass the call to start, otherwise
  // the scheduled cancellation won't be created, and the call will fail early.
  ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>(
      method,
      MoreExecutors.directExecutor(),
      baseCallOptions.withDeadline(Deadline.after(1, TimeUnit.SECONDS)),
      provider,
      deadlineCancellationExecutor,
      channelCallTracer);

  call.start(callListener, new Metadata());

  fakeClock.forwardNanos(TimeUnit.SECONDS.toNanos(1) + 1);

  verify(stream, times(1)).cancel(statusCaptor.capture());
  assertEquals(Status.Code.DEADLINE_EXCEEDED, statusCaptor.getValue().getCode());
}

@Test
public void streamCancelAbortsDeadlineTimer() {
  fakeClock.forwardTime(System.nanoTime(), TimeUnit.NANOSECONDS);

  ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>(
      method,
      MoreExecutors.directExecutor(),
      baseCallOptions.withDeadline(Deadline.after(1, TimeUnit.SECONDS)),
      provider,
      deadlineCancellationExecutor,
      channelCallTracer);
  call.start(callListener, new Metadata());
  call.cancel("canceled", null);

  // Run the deadline timer, which should have been cancelled by the previous call to cancel()
  fakeClock.forwardNanos(TimeUnit.SECONDS.toNanos(1) + 1);

  verify(stream, times(1)).cancel(statusCaptor.capture());

  assertEquals(Status.CANCELLED.getCode(), statusCaptor.getValue().getCode());
}

@Test
public void testDeadlinePropagation() throws Exception {
  final AtomicInteger recursionDepthRemaining = new AtomicInteger(3);
  final SettableFuture<Deadline> finalDeadline = SettableFuture.create();
  class DeadlineSaver extends TestServiceGrpc.TestServiceImplBase {
    @Override
    public void unaryCall(final SimpleRequest request,
        final StreamObserver<SimpleResponse> responseObserver) {
      Context.currentContextExecutor(otherWork).execute(new Runnable() {
        @Override
        public void run() {
          try {
            if (recursionDepthRemaining.decrementAndGet() == 0) {
              finalDeadline.set(Context.current().getDeadline());
              responseObserver.onNext(SimpleResponse.getDefaultInstance());
            } else {
              responseObserver.onNext(blockingStub.unaryCall(request));
            }
            responseObserver.onCompleted();
          } catch (Exception ex) {
            responseObserver.onError(ex);
          }
        }
      });
    }
  }

  server = InProcessServerBuilder.forName("channel").executor(otherWork)
      .addService(new DeadlineSaver())
      .build().start();

  Deadline initialDeadline = Deadline.after(1, TimeUnit.MINUTES);
  blockingStub.withDeadline(initialDeadline).unaryCall(SimpleRequest.getDefaultInstance());
  assertNotSame(initialDeadline, finalDeadline);
  // Since deadline is re-calculated at each hop, some variance is acceptable and expected.
  assertAbout(deadline())
      .that(finalDeadline.get()).isWithin(1, TimeUnit.SECONDS).of(initialDeadline);
}

@Test
public void testConfigureDeadline() {
  Deadline deadline = Deadline.after(2, NANOSECONDS);
  // Create a default stub
  TestServiceGrpc.TestServiceBlockingStub stub = TestServiceGrpc.newBlockingStub(channel);
  assertNull(stub.getCallOptions().getDeadline());
  // Reconfigure it
  TestServiceGrpc.TestServiceBlockingStub reconfiguredStub = stub.withDeadline(deadline);
  // New altered config
  assertEquals(deadline, reconfiguredStub.getCallOptions().getDeadline());
  // Default config unchanged
  assertNull(stub.getCallOptions().getDeadline());
}

/**
 * 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);
}

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

public Value(MessageLite response, Deadline maxAgeDeadline) {
  this.response = response;
  this.maxAgeDeadline = maxAgeDeadline;
}

public static Subject.Factory<DeadlineSubject, Deadline> deadline() {
  return deadlineFactory;
}

private DeadlineSubject(FailureMetadata metadata, Deadline subject) {
  super(metadata, subject);
}

@Override
public DeadlineSubject createSubject(FailureMetadata metadata, Deadline that) {
  return new DeadlineSubject(metadata, that);
}

private ScheduledFuture<?> startDeadlineTimer(Deadline deadline) {
  long remainingNanos = deadline.timeRemaining(TimeUnit.NANOSECONDS);
  return deadlineCancellationExecutor.schedule(
      new LogExceptionRunnable(
          new DeadlineTimer(remainingNanos)), remainingNanos, 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());
}

/**
 * Fails if the subject was expected to be within the tolerance of the given value but was not
 * <i>or</i> if it was expected <i>not</i> to be within the tolerance but was. The expectation,
 * subject, and tolerance are all specified earlier in the fluent call chain.
 */
public abstract void of(Deadline expectedDeadline);

/**
 * Returns a new stub with an absolute deadline.
 *
 * <p>This is mostly used for propagating an existing deadline. {@link #withDeadlineAfter} is the
 * recommended way of setting a new deadline,
 *
 * @since 1.0.0
 * @param deadline the deadline or {@code null} for unsetting the deadline.
 */
public final S withDeadline(@Nullable Deadline deadline) {
  return build(channel, callOptions.withDeadline(deadline));
}