VertxRetryHelper.java

class VertxRetryHelper {
    private static final Logger LOG = LoggerFactory.getLogger(VertxRetryHelper.class);
    private final Vertx vertx;
  
    public VertxRetryHelper(Vertx vertx) {
      this.vertx = vertx;
    }
    
    public <T> Supplier<CompletionStage<T>> decorateCompletionStage(
            RateLimiter rateLimiter,
            Supplier<CompletionStage<T>> supplier) {
        return () -> {
            final CompletableFuture<T> promise = new CompletableFuture<>();
            final BiConsumer<T, Throwable> biConsumer = (result, throwable) -> {
                if (throwable != null) {
                    rateLimiter.onError(throwable);
                    promise.completeExceptionally(throwable);
                } else {
                    rateLimiter.onResult(result);
                    promise.complete(result);
                }
            };
            try {
                long waitMs = rateLimiter.reservePermission() / 1000000;
                if (waitMs == 0) {
                    supplier.get().whenComplete(biConsumer);
                } else if (waitMs > 0) {
                    LOG.warn("Waiting {}ms", waitMs);
                    vertx.setTimer(waitMs, u -> supplier.get().whenComplete(biConsumer));
                } else {
                    promise.completeExceptionally(RequestNotPermitted.createRequestNotPermitted(rateLimiter));
                }
            } catch (Exception exception) {
                rateLimiter.onError(exception);
                promise.completeExceptionally(exception);
            }
            return promise;
        };
    }
}