Java - Spring Boot - Reactive Redis Stream ( TEXT_EVENT_STREAM_VALUE )

Question

I want to write an endpoint which always shows the newest messages of a redis stream (reactive).

---

The entities look like this {'key' : 'some_key', 'status' : 'some_string'}.

So I would like to have the following result:

1. Page is called, content would be for instance displaying an entity:

{'key' : 'abc', 'status' : 'status_A'}

the page is not closed

2. Then a new entity is added to the stream

XADD mystream * key abc status statusB

3. Now I would prefer to see each item of the Stream, without updating the Tab

{'key' : 'abc', 'status' : 'status_A'}
{'key' : 'abc', 'status' : 'status_B'}

---

When I try to mock this behavior it works and I get the expected output.

    @GetMapping(value="/light/live/mock", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @ResponseBody
    public Flux<Light> liveLightMock() {
        List<Light> test = Arrays.asList(new Light("key", "on") , new Light("key", "off"),
                new Light("key", "on") , new Light("key", "off"),
                new Light("key", "on") , new Light("key", "off"),
                new Light("key", "on") , new Light("key", "off"),
                new Light("key", "on") , new Light("key", "off"));
        return Flux.fromIterable(test).delayElements(Duration.ofMillis(500));
    }

The individual elements of the list are displayed one after another with a 500ms Delay between items.

However, when I try to access Redis instead of the mocked variant, it no longer works. I try to test the partial functions successively. So that my idea works first the save (1) function must work, if the save function works, displaying old records without reactiv features must work (2) and last but not least if both work i kinda need to get the reactiv part going.

Maybe you guys can help me get the Reactive Part Working. Im working on it for days without getting any improvements.

Ty guys :)

Test 1) - Saving Function (Short Version)

looks like its working.

    @GetMapping(value="/light/create", produces = MediaType.APPLICATION_JSON_VALUE)
    @ResponseBody
    public Flux<Light> createTestLight() {
        String status = (++statusIdx % 2 == 0) ? "on" : "off";
        Light light = new Light(Consts.LIGHT_ID, status);
        return LightRepository.save(light).flux();
    }

    @Override
    public Mono<Light> save(Light light) {
        Map<String, String> lightMap = new HashMap<>();
        lightMap.put("key", light.getKey());
        lightMap.put("status", light.getStatus());

        return operations.opsForStream(redisSerializationContext)
                .add("mystream", lightMap)
                .map(__ -> light);
    }

Test 2) - Loading/Reading Function (Short Version)

seems to be working, but not reaktiv -> i add a new entity while a WebView was Open, the View showed all Items but didnt Updated once i added new items. after reloading i saw every item

How can i get getLightsto return something that is working with TEXT_EVENT_STREAM_VALUE which subscribes to the stream?

    @Override
    public Flux<Object> getLights() {
        ReadOffset readOffset = ReadOffset.from("0");
        StreamOffset<String> offset = StreamOffset.fromStart("mystream"); //fromStart or Latest

        Function<? super MapRecord<String, Object, Object>, ? extends Publisher<?>> mapFunc = entries -> {
            Map<Object, Object> kvp = entries.getValue();
            String key = (String) kvp.get("key");
            String status = (String) kvp.get("status");
            Light light = new Light(key, status);
            return Flux.just(light);
        };

        return operations.opsForStream()
                .read(offset)
                .flatMap(mapFunc);
    }

    @GetMapping(value="/light/live", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @ResponseBody
    public Flux<Object> lightLive() {
        return LightRepository.getLights();
    }

---

Test 1) - Saving Function (Long Version)

The Endpoint & Saving Functions are part of Diffrent Classes. String status = (++statusIdx % 2 == 0) ? "on" : "off"; flip flops the status from on to off, to on, to off, ...

    @GetMapping(value="/light/create", produces = MediaType.APPLICATION_JSON_VALUE)
    @ResponseBody
    public Flux<Light> createTestLight() {
        String status = (++statusIdx % 2 == 0) ? "on" : "off";
        Light light = new Light(Consts.LIGHT_ID, status);
        return LightRepository.save(light).flux();
    }

    @Override
    public Mono<Light> save(Light light) {
        Map<String, String> lightMap = new HashMap<>();
        lightMap.put("key", light.getKey());
        lightMap.put("status", light.getStatus());

        return operations.opsForStream(redisSerializationContext)
                .add("mystream", lightMap)
                .map(__ -> light);
    }

To Validate the Functions i

1. Delted the Stream, to Empty it

127.0.0.1:6379> del mystream
(integer) 1
127.0.0.1:6379> XLEN myStream
(integer) 0

Called the Creation Endpoint twice /light/create i expected the Stream now to have two Items, on with status = on, and one with off

127.0.0.1:6379> XLEN mystream
(integer) 2
127.0.0.1:6379> xread STREAMS mystream 0-0
1) 1) "mystream"
   2) 1) 1) "1610456865517-0"
         2) 1) "key"
            2) "light_1"
            3) "status"
            4) "off"
      2) 1) "1610456866708-0"
         2) 1) "key"
            2) "light_1"
            3) "status"
            4) "on"

It looks like the Saving part is Working.

Test 2) - Loading/Reading Function (Long Version)

seems to be working, but not reaktiv -> i add a new entity and the page updates its values

    @Override
    public Flux<Object> getLights() {
        ReadOffset readOffset = ReadOffset.from("0");
        StreamOffset<String> offset = StreamOffset.fromStart("mystream"); //fromStart or Latest

        Function<? super MapRecord<String, Object, Object>, ? extends Publisher<?>> mapFunc = entries -> {
            Map<Object, Object> kvp = entries.getValue();
            String key = (String) kvp.get("key");
            String status = (String) kvp.get("status");
            Light light = new Light(key, status);
            return Flux.just(light);
        };

        return operations.opsForStream()
                .read(offset)
                .flatMap(mapFunc);
    }

    @GetMapping(value="/light/live", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @ResponseBody
    public Flux<Object> lightLive() {
        return LightRepository.getLights();
    }

1. Calling /light/live -> i should have N entries -> if I can see Entries, the normal Display is Working (non Reactive)
2. Calling /light/create twice -> the live Few should have added 2 Entries -> N+2Entries
3. Waiting 1 Minute just to be Safe
4. The View Should Show N+2 Entries for the Reactiv Part to be working
5. Refresh View from 1 (/light/live), should still show the same amount if Reactiv Works

Displaying the Information works (1), the Adding part of (2) worked, checked per Terminal, 4) didnt work

ergo the Display is working, but its not reactive

after i refreshed the Browser (5) i got the expected N+2 entries - so (2) worked aswell

---

Answer 1

There's a misconception here, reading from Redis reactively does not mean you have subscribed for new events.

Reactive will not provide you live updates, it will call Redis once and it will display whatever is there. So even if you wait for a day or two nothing is going to change in UI/Console, you will still seeing N entries.

You need to either use Redis PUB/SUB or you need to call Redis repetitively to get the latest update.

EDIT:

A working solution..

  private List<Light> reactiveReadToList() {
    log.info("reactiveReadToList");
    return read().collectList().block();
  }

  private Flux<Light> read() {
    StreamOffset<Object> offset = StreamOffset.fromStart("mystream");
    return redisTemplate
        .opsForStream()
        .read(offset)
        .flatMap(
            e -> {
              Map<Object, Object> kvp = e.getValue();
              String key = (String) kvp.get("key");
              String id = (String) kvp.get("id");
              String status = (String) kvp.get("status");
              Light light = new Light(id, key, status);
              log.info("{}", light);
              return Flux.just(light);
            });
  }

A reader that reads data from Redis on demand using reactive template and send it to the client as it sees using offset, it sends only one event at once we can send all of them.

  @RequiredArgsConstructor
  class DataReader {
    @NonNull FluxSink<Light> sink;
    private List<Light> readLights = null;
    private int currentOffset = 0;
    void register() {
      readLights = reactiveReadToList();
      sink.onRequest(
          e -> {
            long demand = sink.requestedFromDownstream();
            for (int i = 0; i < demand && currentOffset < readLights.size(); i++, currentOffset++) {
              sink.next(readLights.get(currentOffset));
            }
            if (currentOffset == readLights.size()) {
              readLights = reactiveReadToList();
              currentOffset = 0;
            }
          });
    }
  }

A method that uses DataReader to generate flux

  public Flux<Light> getLights() {
    return Flux.create(e -> new DataReader(e).register());
  }

Now we've added an onRequest method on the sink to handle the client demand, this reads data from the Redis stream as required and sends it to the client.

This looks to be very CPU intensive maybe we should delay the calls if there're no more new events, maybe add a sleep call inside register method if we see there're not new elements in the stream.