How do you add elements to a Redis Hash and set the expiration using ReactiveRedisTemplate without block?

Question

If it wasn't reactive I'd do something like.

/**
 * Given a refresh token and the user name, create a "payload" to represent
 * secret data and store it into Redis as a hash and set it to expire in 30 seconds.
 */
Map<String, String> provideAuthenticatedData(String refreshTokenMono, String username) {
  var ops = redisTemplate.opsForHash();
  var payload = Map.of(
    "username", username,
    "secret", UUID.randomUUID().toString()
  );
  var puts = payload.entrySet()
    .stream()
    .map(e->ops.putIfAbsent(refreshToken, e.key(), e.value())
    .filter(success -> !success) // finds those that have failed
    .toList();
  if (!puts.isEmpty()) {
    throw new IllegalStateException("some elements failed to save");
  }
  var expireCheck = redisTemplate.expireAt(refreshToken, Instant.now().plusSeconds(30));
  if (!expireCheck) {
    throw new IllegalStateException("unable to expire");
  }
  return payload;
}

Trying to do it with Reactive it looks to get a bit messier and I got stuck after a point

/**
 * Given a refresh token mono and the user name, create a "payload" to represent
 * secret data and store it into Redis as a hash and set it to expire in 30 seconds.
 */
Mono<Map<String, String>> provideAuthenticatedData(Mono<String> refreshTokenMono, String username) {
  var ops = reactiveRedisTemplate.opsForHash();
  var payload = Map.of(
    "username", username,
    "secret", UUID.randomUUID().toString()
  );
  return refreshTokenMono
    .flatMapIterable(
      refreshToken -> payload.entrySet()
        .stream()
        .map(
          e -> ops.putIfAbsent(refreshToken, e.getKey(), e.getValue())
        )
        .toList() // can't find an operator that would take a stream
    )
    // at this point I have a Flux<Mono<Boolean>>
    // somehow I have to find out if any of them are false then return a Mono.error()
    // then once all of it is done, set the key to expire
    // finally return the payload I originally created
}

Another approach I did was this but it does not do any error handling.

Mono<Map<String, String>> provideAuthenticatedDataMono(
      Mono<String> refreshTokenMono, String username) {

  var ops = reactiveRedisTemplate.opsForHash();
  var payload = Map.of(
    "username", username,
    "secret", UUID.randomUUID().toString()
  );

  return refreshTokenMono
    .doOnNext(
        refreshToken ->
          payload
            .entrySet()
            .stream()
            .map(
              e ->  ops.putIfAbsent(
                refreshToken,
                e.getKey(),
                e.getValue())
             )
            .forEach(Mono::subscribe)
    )
    .doOnNext(
       refreshToken ->
         redisTemplate
           .expireAt(
             refreshToken,
             Instant.now().plusSeconds(30)
           )
           .subscribe()
    )
    .flatMap((x) -> just(payload));
  }
}

Answer 1

The main idea of reactive is to work in stream so you should avoid subscribing everywhere you should just return the Flux, Mono Stream.

THe first example is not working as you dont subscribe the mono that redis gives you.

As you are using reactive why you mix it with java stream.

one solution would be like

  public Mono< YOUR_OBJECT > save(Object YOUR_OBJECT) {
return template.opsForValue().set(YOUR_OBJECT.key, YOUR_OBJECT)
    .filter(aBoolean -> aBoolean)
    .map(aBoolean -> YOUR_OBJECT)
    .switchIfEmpty(Mono.error(new RuntimeException("Could not save data to redis")));

}

And you should be continuing the stream till the end when it will be subscribed by a controller or you