@Test @SuppressWarnings({"unchecked"}) void retryWithDelay() { Timer timer = Timer.createAndStart(CoreEnvironment.DEFAULT_MAX_NUM_REQUESTS_IN_RETRY); RetryStrategy retryStrategy = mock(RetryStrategy.class); when(retryStrategy.shouldRetry(any(Request.class), any(RetryReason.class))).thenReturn( CompletableFuture.completedFuture(RetryAction.withDuration(Duration.ofMillis(200))) ); Request<?> request = mock(Request.class); RequestContext requestContext = mock(RequestContext.class); when(request.completed()).thenReturn(false); when(request.context()).thenReturn(requestContext); when(request.retryStrategy()).thenReturn(retryStrategy); when(request.absoluteTimeout()).thenReturn(System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(2500)); Core core = mock(Core.class); CoreEnvironment env = mock(CoreEnvironment.class); SimpleEventBus eventBus = new SimpleEventBus(true); when(env.timer()).thenReturn(timer); when(env.eventBus()).thenReturn(eventBus); CoreContext ctx = new CoreContext(core, 1, env, mock(Authenticator.class)); long start = System.nanoTime(); RetryOrchestrator.maybeRetry(ctx, request, RetryReason.UNKNOWN); verify(requestContext, times(1)) .incrementRetryAttempts(Duration.ofMillis(200), RetryReason.UNKNOWN); verify(request, never()).cancel(CancellationReason.noMoreRetries(RetryReason.UNKNOWN)); waitUntilCondition(() -> !Mockito.mockingDetails(core).getInvocations().isEmpty()); long end = System.nanoTime(); verify(core, times(1)).send(request, false); verify(core, never()).send(request); assertTrue(TimeUnit.NANOSECONDS.toMillis(end - start) >= 200); timer.stop(); assertEquals(1, eventBus.publishedEvents().size()); RequestRetryScheduledEvent retryEvent = (RequestRetryScheduledEvent) eventBus.publishedEvents().get(0); assertEquals(Event.Severity.DEBUG, retryEvent.severity()); assertEquals(Event.Category.REQUEST.path(), retryEvent.category()); assertEquals(requestContext, retryEvent.context()); assertEquals(RetryReason.UNKNOWN, retryEvent.retryReason()); }
public static void maybeRetry(final CoreContext ctx, final Request<? extends Response> request, final RetryReason reason) { if (request.completed()) { return; } if (reason.alwaysRetry()) { retryWithDuration(ctx, request, controlledBackoff(request.context().retryAttempts()), reason); return; } request.retryStrategy().shouldRetry(request, reason).whenComplete((retryAction, throwable) -> { if (throwable != null) { ctx.environment().eventBus().publish( new RequestNotRetriedEvent(Event.Severity.INFO, request.getClass(), request.context(), reason, throwable) ); } Optional<Duration> duration = retryAction.duration(); if (duration.isPresent()) { final Duration cappedDuration = capDuration(duration.get(), request); retryWithDuration(ctx, request, cappedDuration, reason); } else { Event.Severity severity = request instanceof UnmonitoredRequest ? Event.Severity.DEBUG : Event.Severity.INFO; ctx.environment().eventBus().publish( new RequestNotRetriedEvent(severity, request.getClass(), request.context(), reason, null) ); request.cancel(CancellationReason.noMoreRetries(reason)); } }); }
RetryOrchestrator { public static void maybeRetry(final CoreContext ctx, final Request<? extends Response> request, final RetryReason reason) { if (request.completed()) { return; } if (reason.alwaysRetry()) { retryWithDuration(ctx, request, controlledBackoff(request.context().retryAttempts()), reason); return; } request.retryStrategy().shouldRetry(request, reason).whenComplete((retryAction, throwable) -> { if (throwable != null) { ctx.environment().eventBus().publish( new RequestNotRetriedEvent(Event.Severity.INFO, request.getClass(), request.context(), reason, throwable) ); } Optional<Duration> duration = retryAction.duration(); if (duration.isPresent()) { final Duration cappedDuration = capDuration(duration.get(), request); retryWithDuration(ctx, request, cappedDuration, reason); } else { Event.Severity severity = request instanceof UnmonitoredRequest ? Event.Severity.DEBUG : Event.Severity.INFO; ctx.environment().eventBus().publish( new RequestNotRetriedEvent(severity, request.getClass(), request.context(), reason, null) ); request.cancel(CancellationReason.noMoreRetries(reason)); } }); } }
RetryOrchestrator { public static void maybeRetry(final CoreContext ctx, final Request<? extends Response> request, final RetryReason reason) { if (request.completed()) { return; } if (reason.alwaysRetry()) { retryWithDuration(ctx, request, controlledBackoff(request.context().retryAttempts()), reason); return; } request.retryStrategy().shouldRetry(request, reason).whenComplete((retryAction, throwable) -> { if (throwable != null) { ctx.environment().eventBus().publish( new RequestNotRetriedEvent(Event.Severity.INFO, request.getClass(), request.context(), reason, throwable) ); } Optional<Duration> duration = retryAction.duration(); if (duration.isPresent()) { final Duration cappedDuration = capDuration(duration.get(), request); retryWithDuration(ctx, request, cappedDuration, reason); } else { Event.Severity severity = request instanceof UnmonitoredRequest ? Event.Severity.DEBUG : Event.Severity.INFO; ctx.environment().eventBus().publish( new RequestNotRetriedEvent(severity, request.getClass(), request.context(), reason, null) ); request.cancel(CancellationReason.noMoreRetries(reason)); } }); }  }
RetryOrchestrator { public static void maybeRetry(final CoreContext ctx, final Request<? extends Response> request, final RetryReason reason) { if (request.completed()) { return; } if (reason.alwaysRetry()) { retryWithDuration(ctx, request, controlledBackoff(request.context().retryAttempts()), reason); return; } request.retryStrategy().shouldRetry(request, reason).whenComplete((retryAction, throwable) -> { if (throwable != null) { ctx.environment().eventBus().publish( new RequestNotRetriedEvent(Event.Severity.INFO, request.getClass(), request.context(), reason, throwable) ); } Optional<Duration> duration = retryAction.duration(); if (duration.isPresent()) { final Duration cappedDuration = capDuration(duration.get(), request); retryWithDuration(ctx, request, cappedDuration, reason); } else { Event.Severity severity = request instanceof UnmonitoredRequest ? Event.Severity.DEBUG : Event.Severity.INFO; ctx.environment().eventBus().publish( new RequestNotRetriedEvent(severity, request.getClass(), request.context(), reason, null) ); request.cancel(CancellationReason.noMoreRetries(reason)); } }); }  static void maybeRetry(final CoreContext ctx, final Request<? extends Response> request,
                                final RetryReason reason); @Stability.Internal static Duration capDuration(final Duration uncappedDuration, final Request<? extends Response> request); }
RetryOrchestrator { public static void maybeRetry(final CoreContext ctx, final Request<? extends Response> request, final RetryReason reason) { if (request.completed()) { return; } if (reason.alwaysRetry()) { retryWithDuration(ctx, request, controlledBackoff(request.context().retryAttempts()), reason); return; } request.retryStrategy().shouldRetry(request, reason).whenComplete((retryAction, throwable) -> { if (throwable != null) { ctx.environment().eventBus().publish( new RequestNotRetriedEvent(Event.Severity.INFO, request.getClass(), request.context(), reason, throwable) ); } Optional<Duration> duration = retryAction.duration(); if (duration.isPresent()) { final Duration cappedDuration = capDuration(duration.get(), request); retryWithDuration(ctx, request, cappedDuration, reason); } else { Event.Severity severity = request instanceof UnmonitoredRequest ? Event.Severity.DEBUG : Event.Severity.INFO; ctx.environment().eventBus().publish( new RequestNotRetriedEvent(severity, request.getClass(), request.context(), reason, null) ); request.cancel(CancellationReason.noMoreRetries(reason)); } }); }  static void maybeRetry(final CoreContext ctx, final Request<? extends Response> request,
                                final RetryReason reason); @Stability.Internal static Duration capDuration(final Duration uncappedDuration, final Request<? extends Response> request);  }