how can i call this class and apply this filter with reactive redis, from my restcontroller? without using spring cloud gateway:
package com.mediaworld.apimanager.configuration.redis;

import org.reactivestreams.Publisher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.core.ReactiveRedisCallback;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.serializer.GenericToStringSerializer;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.web.reactive.function.server.HandlerFilterFunction;
import org.springframework.web.reactive.function.server.HandlerFunction;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Mono;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.time.LocalTime;
import java.util.List;
import java.util.Optional;

import static org.springframework.http.HttpStatus.TOO_MANY_REQUESTS;

public class RateLimiterHandlerFilterFunction implements HandlerFilterFunction<ServerResponse,ServerResponse> {

    @Value("${MAX_REQUESTS_PER_MINUTE}")
    private static Long MAX_REQUESTS_PER_MINUTE = 20L;
    private ReactiveRedisTemplate<String, Long> redisTemplate;

    public RateLimiterHandlerFilterFunction(ReactiveRedisTemplate<String, Long> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Override
    public Mono<ServerResponse> filter(ServerRequest request, HandlerFunction<ServerResponse> next) {
        int currentMinute = LocalTime.now().getMinute();
        String key = String.format("rl_%s:%s", requestAddress(request.remoteAddress()), currentMinute);

        return redisTemplate //
                .opsForValue().get(key) //
                .flatMap( //
                        value -> value >= MAX_REQUESTS_PER_MINUTE ? //
                                ServerResponse.status(TOO_MANY_REQUESTS).build() : //
                                incrAndExpireKey(key, request, next) //
                ).switchIfEmpty(incrAndExpireKey(key, request, next));
    }

    @Bean
    ReactiveRedisTemplate<String, Long> reactiveRedisTemplate(ReactiveRedisConnectionFactory factory) {
        JdkSerializationRedisSerializer jdkSerializationRedisSerializer = new JdkSerializationRedisSerializer();
        StringRedisSerializer stringRedisSerializer = StringRedisSerializer.UTF_8;
        GenericToStringSerializer<Long> longToStringSerializer = new GenericToStringSerializer<>(Long.class);
        ReactiveRedisTemplate<String, Long> template = new ReactiveRedisTemplate<>(factory,
                RedisSerializationContext.<String, Long>newSerializationContext(jdkSerializationRedisSerializer)
                        .key(stringRedisSerializer).value(longToStringSerializer).build());
        return template;
    }

    private String requestAddress(Optional<InetSocketAddress> maybeAddress) {
        return maybeAddress.isPresent() ? maybeAddress.get().getHostName() : "";
    }


    /**
     *
     Now we need to implement the logic to execute the INCR and an EXPIRE logic outlined in Fixed Window implementation using Spring Data Redis Reactive:

     MULTI
     INCR [user-api-key]:[current minute number]
     EXPIRE [user-api-key]:[current minute number] 59
     EXEC

     The "Basic Rate Limiting" recipe calls for the use of a Redis Transaction in which the commands are sent to the server, accumulated in serial way and executed sequentially without any possible interruption by a request from another client.

     Basically, we want the INCR and EXPIRE calls to update the requests-per-unit-of-time counter to happen atomically or not at all. Given that a Reactive API and Redis Transactions (MULTI/EXEC) are not compatible paradigms, which boils down to "you cannot listen to a command that is going to be executed in the future" in a chain of reactive commands A "best possible" approach to achieve this behavior with a reactive API is by using the ReactiveRedisTemplate execute method which takes a ReactiveRedisCallback guaranteing that at least the commands will run on the same Redis connection, but, this is by no means a real "transaction":
     */
    private Mono<ServerResponse> incrAndExpireKey(String key, ServerRequest request,
                                                  HandlerFunction<ServerResponse> next) {
        return redisTemplate.execute(new ReactiveRedisCallback<List<Object>>() {
            @Override
            public Publisher<List<Object>> doInRedis(ReactiveRedisConnection connection) throws DataAccessException {
                ByteBuffer bbKey = ByteBuffer.wrap(key.getBytes());
                return Mono.zip( //
                        connection.numberCommands().incr(bbKey), //
                        connection.keyCommands().expire(bbKey, Duration.ofSeconds(59L)) //
                ).then(Mono.empty());
            }
        }).then(next.handle(request));
    }


}