@Bean
public ReactiveRedisTemplate<Object, Object> reactiveRedisTemplate(
    ReactiveRedisConnectionFactory reactiveRedisConnectionFactory, ResourceLoader resourceLoader) {

    FstSerializationRedisSerializer serializer = new FstSerializationRedisSerializer(() -> {
        FSTConfiguration configuration = FSTConfiguration.createDefaultConfiguration()
            .setForceSerializable(true);
        configuration.setClassLoader(resourceLoader.getClassLoader());
        return configuration;
    });
    @SuppressWarnings("all")
    RedisSerializationContext<Object, Object> serializationContext = RedisSerializationContext
        .newSerializationContext()
        .key((RedisSerializer)new StringRedisSerializer())
        .value(serializer)
        .hashKey(StringRedisSerializer.UTF_8)
        .hashValue(serializer)
        .build();

    return new ReactiveRedisTemplate<>(reactiveRedisConnectionFactory, serializationContext);
}
 

@Override
public void handlerPlugin(final PluginData pluginData) {
    if (Objects.nonNull(pluginData) && pluginData.getEnabled()) {
        //init redis
        RateLimiterConfig rateLimiterConfig = GsonUtils.getInstance().fromJson(pluginData.getConfig(), RateLimiterConfig.class);
        //spring data redisTemplate
        if (Objects.isNull(Singleton.INST.get(ReactiveRedisTemplate.class))
                || Objects.isNull(Singleton.INST.get(RateLimiterConfig.class))
                || !rateLimiterConfig.equals(Singleton.INST.get(RateLimiterConfig.class))) {
            LettuceConnectionFactory lettuceConnectionFactory = createLettuceConnectionFactory(rateLimiterConfig);
            lettuceConnectionFactory.afterPropertiesSet();
            RedisSerializer<String> serializer = new StringRedisSerializer();
            RedisSerializationContext<String, String> serializationContext =
                    RedisSerializationContext.<String, String>newSerializationContext().key(serializer).value(serializer).hashKey(serializer).hashValue(serializer).build();
            ReactiveRedisTemplate<String, String> reactiveRedisTemplate = new ReactiveRedisTemplate<>(lettuceConnectionFactory, serializationContext);
            Singleton.INST.single(ReactiveRedisTemplate.class, reactiveRedisTemplate);
            Singleton.INST.single(RateLimiterConfig.class, rateLimiterConfig);
        }
    }
}
 

/**
 * This uses a basic token bucket algorithm and relies on the fact that Redis scripts
 * execute atomically. No other operations can run between fetching the count and
 * writing the new count.
 *
 * @param id            is rule id
 * @param replenishRate replenishRate
 * @param burstCapacity burstCapacity
 * @return {@code Mono<Response>} to indicate when request processing is complete
 */
@SuppressWarnings("unchecked")
public Mono<RateLimiterResponse> isAllowed(final String id, final double replenishRate, final double burstCapacity) {
    if (!this.initialized.get()) {
        throw new IllegalStateException("RedisRateLimiter is not initialized");
    }
    List<String> keys = getKeys(id);
    List<String> scriptArgs = Arrays.asList(replenishRate + "", burstCapacity + "", Instant.now().getEpochSecond() + "", "1");
    Flux<List<Long>> resultFlux = Singleton.INST.get(ReactiveRedisTemplate.class).execute(this.script, keys, scriptArgs);
    return resultFlux.onErrorResume(throwable -> Flux.just(Arrays.asList(1L, -1L)))
            .reduce(new ArrayList<Long>(), (longs, l) -> {
                longs.addAll(l);
                return longs;
            }).map(results -> {
                boolean allowed = results.get(0) == 1L;
                Long tokensLeft = results.get(1);
                RateLimiterResponse rateLimiterResponse = new RateLimiterResponse(allowed, tokensLeft);
                log.info("RateLimiter response:{}", rateLimiterResponse.toString());
                return rateLimiterResponse;
            }).doOnError(throwable -> {
                log.error("Error determining if user allowed from redis:{}", throwable.getMessage());
            });
}
 

@Test
@SuppressWarnings("unused")
void newReactiveRedisSessionRepository() {
	LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory();
	RedisSerializationContext<String, Object> serializationContext = RedisSerializationContext
			.<String, Object>newSerializationContext(new JdkSerializationRedisSerializer()).build();

	// tag::new-reactiveredissessionrepository[]
	// ... create and configure connectionFactory and serializationContext ...

	ReactiveRedisTemplate<String, Object> redisTemplate = new ReactiveRedisTemplate<>(connectionFactory,
			serializationContext);

	ReactiveSessionRepository<? extends Session> repository = new ReactiveRedisSessionRepository(redisTemplate);
	// end::new-reactiveredissessionrepository[]
}
 

@Bean
public ReactiveRedisTemplate<String, String> stringReactiveRedisTemplate(ReactiveRedisConnectionFactory connectionFactory){
    RedisSerializationContext<String, String> serializationContext = RedisSerializationContext
        .<String, String>newSerializationContext(new StringRedisSerializer())
        .hashKey(new StringRedisSerializer())
        .hashValue(new StringRedisSerializer())
        .build();
    return new ReactiveRedisTemplate<>(connectionFactory, serializationContext);
}
 

public RateCheckRedisRateLimiter(ReactiveRedisTemplate<String, String> redisTemplate,
                                 RedisScript<List<Long>> script, Validator validator) {
    super(Config.class, CONFIGURATION_PROPERTY_NAME, validator);
    this.redisTemplate = redisTemplate;
    this.script = script;
    initialized.compareAndSet(false, true);
}
 

@Override
@SuppressWarnings("unchecked")
public void setApplicationContext(ApplicationContext context) throws BeansException {
    if (initialized.compareAndSet(false, true)) {
        this.redisTemplate = context.getBean("stringReactiveRedisTemplate", ReactiveRedisTemplate.class);
        this.script = context.getBean(REDIS_SCRIPT_NAME, RedisScript.class);
        if (context.getBeanNamesForType(Validator.class).length > 0) {
            this.setValidator(context.getBean(Validator.class));
        }
    }
}
 

@Override
  public void initialize(GenericApplicationContext context) {
RedisReactiveAutoConfiguration redisAutoConfiguration = new RedisReactiveAutoConfiguration();
context.registerBean("reactiveRedisTemplate", ReactiveRedisTemplate.class,
		() -> redisAutoConfiguration.reactiveRedisTemplate(context.getBean(ReactiveRedisConnectionFactory.class), context),
		(definition) -> ((RootBeanDefinition) definition).setTargetType(ResolvableType.forClassWithGenerics(ReactiveRedisTemplate.class, Object.class, Object.class)));
context.registerBean("reactiveStringRedisTemplate", ReactiveStringRedisTemplate.class,
		() -> redisAutoConfiguration.reactiveStringRedisTemplate(context.getBean(ReactiveRedisConnectionFactory.class)));
  }
 

@Bean
public ReactiveRedisTemplate<String, String> stringReactiveRedisTemplate(ReactiveRedisConnectionFactory connectionFactory){
    RedisSerializationContext<String, String> serializationContext = RedisSerializationContext
        .<String, String>newSerializationContext(new StringRedisSerializer())
        .hashKey(new StringRedisSerializer())
        .hashValue(new StringRedisSerializer())
        .build();
    return new ReactiveRedisTemplate<>(connectionFactory, serializationContext);
}
 

@Bean
public ReactiveRedisTemplate<String, Post> reactiveJsonPostRedisTemplate(
    ReactiveRedisConnectionFactory connectionFactory) {

    RedisSerializationContext<String, Post> serializationContext = RedisSerializationContext
        .<String, Post>newSerializationContext(new StringRedisSerializer())
        .hashKey(new StringRedisSerializer())
        .hashValue(new Jackson2JsonRedisSerializer<>(Post.class))
        .build();


    return new ReactiveRedisTemplate<>(connectionFactory, serializationContext);
}
 

@Bean
public ReactiveRedisTemplate<String, Book> reactiveJsonBookRedisTemplate(
		ReactiveRedisConnectionFactory reactiveRedisConnectionFactory) {

	RedisSerializationContext<String, Book> serializationContext = RedisSerializationContext
			.<String, Book>newSerializationContext(new StringRedisSerializer()).hashKey(new StringRedisSerializer())
			.hashValue(new Jackson2JsonRedisSerializer<>(Book.class)).build();

	return new ReactiveRedisTemplate<>(reactiveRedisConnectionFactory, serializationContext);
}
 

/**
 * Configures a {@link ReactiveRedisTemplate} with {@link String} keys and a typed
 * {@link Jackson2JsonRedisSerializer}.
 */
@Bean
public ReactiveRedisTemplate<String, Person> reactiveJsonPersonRedisTemplate(
		ReactiveRedisConnectionFactory connectionFactory) {

	Jackson2JsonRedisSerializer<Person> serializer = new Jackson2JsonRedisSerializer<>(Person.class);
	RedisSerializationContextBuilder<String, Person> builder = RedisSerializationContext
			.newSerializationContext(new StringRedisSerializer());

	RedisSerializationContext<String, Person> serializationContext = builder.value(serializer).build();

	return new ReactiveRedisTemplate<>(connectionFactory, serializationContext);
}
 

/**
 * Configures a {@link ReactiveRedisTemplate} with {@link String} keys and {@link GenericJackson2JsonRedisSerializer}.
 */
@Bean
public ReactiveRedisTemplate<String, Object> reactiveJsonObjectRedisTemplate(
		ReactiveRedisConnectionFactory connectionFactory) {

	RedisSerializationContextBuilder<String, Object> builder = RedisSerializationContext
			.newSerializationContext(new StringRedisSerializer());

	RedisSerializationContext<String, Object> serializationContext = builder
			.value(new GenericJackson2JsonRedisSerializer("_type")).build();

	return new ReactiveRedisTemplate<>(connectionFactory, serializationContext);
}
 

private ReactiveRedisTemplate<String, Object> createReactiveRedisTemplate() {
	RedisSerializer<String> keySerializer = new StringRedisSerializer();
	RedisSerializer<Object> defaultSerializer = (this.defaultRedisSerializer != null) ? this.defaultRedisSerializer
			: new JdkSerializationRedisSerializer(this.classLoader);
	RedisSerializationContext<String, Object> serializationContext = RedisSerializationContext
			.<String, Object>newSerializationContext(defaultSerializer).key(keySerializer).hashKey(keySerializer)
			.build();
	return new ReactiveRedisTemplate<>(this.redisConnectionFactory, serializationContext);
}
 

@Bean
public ReactiveRedisTemplate<String, Employee> reactiveRedisTemplate(ReactiveRedisConnectionFactory factory) {
    Jackson2JsonRedisSerializer<Employee> serializer = new Jackson2JsonRedisSerializer<>(Employee.class);
    RedisSerializationContext.RedisSerializationContextBuilder<String, Employee> builder = RedisSerializationContext.newSerializationContext(new StringRedisSerializer());
    RedisSerializationContext<String, Employee> context = builder.value(serializer)
        .build();
    return new ReactiveRedisTemplate<>(factory, context);
}
 

@Bean(initMethod = "startup")
public RedisClusterManager clusterManager(JetLinksProperties properties, ReactiveRedisTemplate<Object, Object> template) {
    return new RedisClusterManager(properties.getClusterName(), properties.getServerId(), template);
}
 

@Bean
public ReactiveRedisTemplate<String,String> reactiveRedisTemplate(ReactiveRedisConnectionFactory reactiveRedisConnectionFactory){
  return new ReactiveRedisTemplate<>(reactiveRedisConnectionFactory, RedisSerializationContext.string());
}
 

public FareRepository(ReactiveRedisTemplate<String, String> reactiveRedisTemplate,
    ObjectMapper mapper) {
  this.reactiveRedisTemplate = reactiveRedisTemplate;
  this.mapper = mapper;
}
 

@Bean
public ReactiveRedisTemplate<String,String> reactiveRedisTemplate(ReactiveRedisConnectionFactory reactiveRedisConnectionFactory){
  return new ReactiveRedisTemplate<>(reactiveRedisConnectionFactory, RedisSerializationContext.string());
}
 

public FareRepository(ReactiveRedisTemplate<String, String> reactiveRedisTemplate,
    ObjectMapper mapper) {
  this.reactiveRedisTemplate = reactiveRedisTemplate;
  this.mapper = mapper;
}
 

public TaxiService(ReactiveRedisTemplate<String, String> reactiveRedisTemplate, TaxiRepository taxiRepository) {
    this.reactiveRedisTemplate = reactiveRedisTemplate;
    this.taxiRepository = taxiRepository;
}
 

@Bean
public ReactiveRedisTemplate<String, String> reactiveRedisTemplate(ReactiveRedisConnectionFactory connectionFactory) {
    return new ReactiveRedisTemplate<>(connectionFactory, RedisSerializationContext.string());
}
 

public TaxiBookingService(RedisTemplate<String, String> redisTemplate, ReactiveRedisTemplate<String, String> reactiveRedisTemplate, TaxiBookingRepository taxiBookingRepository) {
    this.redisTemplate = redisTemplate;
    this.reactiveRedisTemplate = reactiveRedisTemplate;
    this.taxiBookingRepository = taxiBookingRepository;
}
 

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

public DefaultRedisRateLimiter(ReactiveRedisTemplate<String, String> redisTemplate, RedisScript<List<Long>> script, Validator validator) {
    super(redisTemplate, script, validator);
}
 

@Bean
public ReactiveRedisTemplate<String,String> reactiveRedisTemplate(ReactiveRedisConnectionFactory reactiveRedisConnectionFactory){
  return new ReactiveRedisTemplate<>(reactiveRedisConnectionFactory, RedisSerializationContext.string());
}
 

public FareRepository(ReactiveRedisTemplate<String, String> reactiveRedisTemplate,
    ObjectMapper mapper) {
  this.reactiveRedisTemplate = reactiveRedisTemplate;
  this.mapper = mapper;
}
 

@Bean
public ReactiveRedisTemplate<String,String> reactiveRedisTemplate(ReactiveRedisConnectionFactory reactiveRedisConnectionFactory){
  return new ReactiveRedisTemplate<>(reactiveRedisConnectionFactory, RedisSerializationContext.string());
}
 

public FareRepository(ReactiveRedisTemplate<String, String> reactiveRedisTemplate,
    ObjectMapper mapper) {
  this.reactiveRedisTemplate = reactiveRedisTemplate;
  this.mapper = mapper;
}
 

public DefaultRedisRateLimiter(ReactiveRedisTemplate<String, String> redisTemplate,
                               RedisScript<List<Long>> script,
                               @Qualifier("defaultValidator") Validator validator) {
    super(redisTemplate, script, validator);
}