asynchronously reserve permission

MyVerticle.java

class MyVerticle extends AbstractVerticle {

    /* Try to acquire permits asynchronously */
    private <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 waitNano = rateLimiter.reservePermission();
                if (waitNano == 0) {
                    supplier.get().whenComplete(biConsumer);
                } else if (waitNano > 0) {
                    long waitMs = waitNano / 1000000;
                    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;
        };
    }
}