Reactive Redis does not continually publish changes to the Flux

user_1091202 0
Viewed 906

I am trying to get live updates on my redis ordered list without success. It seems like it fetches all the items and just ends on the last item. I would like the client to keep get updates upon a new order in my ordered list. What am I missing?

This is my code: 

@RestController
class LiveOrderController {

    @Autowired
    lateinit var redisOperations: ReactiveRedisOperations<String, LiveOrder>

    @GetMapping(produces = [MediaType.TEXT_EVENT_STREAM_VALUE], value = "/orders")
    fun getLiveOrders(): Flux<LiveOrder> {
        val zops = redisOperations?.opsForZSet()
        return zops?.rangeByScore("orders", Range.unbounded())
    }
}
spring-webflux spring-boot kotlin redis reactive-programming
1 Answers

There is no such feature in Redis. First, reactive retrieval of a sorted set is just getting a snapshot, but your calls are going in a reactive fashion. So you need a subscription instead.

If you opt in for keyspace notifications like this (K - enable keyspace notifications, z - include zset commands) :

config set notify-keyspace-events Kz

And subscribe to them in your service like this:

  ReactiveRedisMessageListenerContainer reactiveRedisMessages;
  // ...
  reactiveRedisMessages.receive(new PatternTopic("__keyspace@0__:orders"))
      .map(m -> {
        System.out.println(m);
        return m;
      })
      <further processing>

You would see messages like this: PatternMessage{channel=__keyspace@0__:orders, pattern=__keyspace@0__:orders, message=zadd}. It will notify you that something has been added. And you can react on this somehow - get the full set again, or only some part (head/tail). You might even remember the previous set, get the new one and send the diff.

But what I would really suggest is rearchitecting the flow in some way to use Redis Pub/Sub functionality directly. For example: publisher service instead of directly calling zadd will call eval, which will issue 2 commands: zadd orders 1 x and publish orders "1:x" (any custom message you want, maybe JSON).

Then in your code you will subscribe to your custom topic like this:

return reactiveRedisMessages.receive(new PatternTopic("orders"))
      .map(LiveOrder::fromNotification);
user_785690
Related