Scaling reactive APIs - www.razorcodes.com

code8_RedisPuller3.java

private final Map<String,EventStream> streamsList = new ConcurrentHashMap<>();

public Flux<CustomerUpdatedEvent> getStreamEvents(String eventStreamId, long fromMs) {
    //add new eventStream to Set
    EventStream eventStream = new EventStream(eventStreamId, fromMs, 0, new HashSet());
    streamsList.putIfAbsent(eventStreamId, eventStream);

    //create a Flux and add the Sink to the EventStream to be used for publishing
    Flux<Object> events = Flux.create(sink -> eventStream.addStream(sink));

    //configure the Flux and return it
    return events
            .cast(CustomerUpdatedEvent.class)
            .timeout(Duration.ofMinutes(60))
            .doFinally(e -> {
                eventStream.removeStream(eventStream);
                log.debug("Finally UNRegistering consumer:{}. Cause:{}.", eventStreamId, e);
            });
}