Add support for bucket4j request rate limiter

[spencergibb]

https://github.com/vladimir-bukhtoyarov/bucket4j/blob/master/doc-pages/asynchronous.md

Might be easier to demo.

[vladimir-bukhtoyarov]

It would be interesting to look at results. Feel free to ask any question according to bucket4j in comments here, or in the dedicated gitter chat.

[spencergibb]

Maybe resilience4j would be a better fit.

[rLitto]

Any notes on this? I have talked with the resilience4j team, and their ratelimiter does not support a use case for a ratelimiter shared by multiple instances (eg Gateway scenarios) Another possible option instead is: https://github.com/mokies/ratelimitj

[giger85]

The rate limit filter based on bucket4j was introduced in spring cloud mvc gateway. I have executed stress test using bucket4j + redis.

If some keys are high contention, then request per second (RPS) of server does not scale up. And I found related issue.

• https://github.com/bucket4j/bucket4j/issues/407

I hope this helps. @spencergibb

[NadChel]

I once wrote a simplistic WebFilter to that effect (see below). It could be turned into an OOB GatewayFilter (with some tweaks). We could also have some default @OnMissingBean BucketResolver in GatewayAutoConfiguration. @spencergibb does it sound like a good idea to you?

package com.example.reactivecrptapi.filter;

import com.example.reactivecrptapi.service.bucketResolver.BucketResolver;
import io.github.bucket4j.Bucket;
import io.github.bucket4j.ConsumptionProbe;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebFilter;
import org.springframework.web.server.WebFilterChain;
import reactor.core.publisher.Mono;
import org.apache.commons.lang3.time.DurationFormatUtils;

import java.nio.charset.StandardCharsets;
import java.text.MessageFormat;
import java.time.Duration;

/**
 * A {@link WebFilter} that applies a time limiting strategy to incoming requests
 */
@Component
public class RateLimitingWebFilter implements WebFilter {
    private final BucketResolver bucketResolver;

    public RateLimitingWebFilter(BucketResolver bucketResolver) {
        this.bucketResolver = bucketResolver;
    }

    /**
     * Uses the injected {@link BucketResolver} to obtain a {@link Bucket}
     * matching the exchange and then attempts to consume one token from the bucket
     * and pass the exchange down the filter chain. If no token is available, sets
     * the {@code 429 Too many requests} response status and completes the exchange
     */
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        return bucketResolver.resolveBucket(exchange)
                .map(bucket -> bucket.tryConsumeAndReturnRemaining(1))
                .flatMap(probe -> probe.isConsumed() ?
                        chain.filter(exchange) :
                        write429Response(exchange, probe));
    }

    private Mono<Void> write429Response(ServerWebExchange exchange, ConsumptionProbe probe) {
        ServerHttpResponse response = exchange.getResponse();
        response.setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
        return writeErrorResponseBody(response, probe);
    }

    private Mono<Void> writeErrorResponseBody(ServerHttpResponse response, ConsumptionProbe probe) {
        Duration timeToRefill = Duration.ofNanos(probe.getNanosToWaitForRefill());
        String humanReadableTimeToRefill = makeReadable(timeToRefill);
        String message = MessageFormat.format(
                "Too many requests. Please wait for at least {0} and make another attempt",
                humanReadableTimeToRefill
        );
        DataBuffer bodyDataBuffer = DefaultDataBufferFactory.sharedInstance
                .wrap(message.getBytes(StandardCharsets.UTF_8));
        return response.writeWith(Mono.just(bodyDataBuffer));
    }

    private String makeReadable(Duration duration) {
        return DurationFormatUtils.formatDurationWords(
                duration.toMillis(), true, true);
    }
}