razorcd

@RestController
public class CustomerController {
    private final RedisPuller redisPuller;

    @GetMapping(value="/customer/{customerId}",produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<CustomerUpdatedEvent> getCustomerData(@PathVariable String customerId) {
        return redisPuller.getStreamEvents(customerId, Instant.now().toEpochMilli());
    }
}

razorcd

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));

razorcd

public class RedisPuller implements AutoCloseable {
    
    private final RedisTemplate<String,String> redisTemplate;
    private final Map<String,EventStream> streamsList = new ConcurrentHashMap<>();
    private final Thread pullerThread = getPullerThread(this.streamsList);

    public RedisPuller(RedisTemplate<String,String> redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.pullerThread.start();
    }

razorcd

@RequiredArgsConstructor
public class RedisPuller {
    
    private final RedisTemplate<String,String> redisTemplate;
    private final Map<String,EventStream> streamsList = new ConcurrentHashMap<>();

    public void runPuller() {
        //create stream array of StreamOffset objects
        Set<StreamOffset<String>> streams = buildStreamOffsets(this.streamsList.values().stream());
        StreamOffset<String>[] streamArray = new StreamOffset[streams.size()];

razorcd

@Value
public class EventStream {
    private final String eventStreamId;
    private final Long offsetTimeMs;
    private final Long offsetCount;
    private final Set<FluxSink<Object>> fluxSinks;

    public void addStream(FluxSink<Object> newFluxSinkObject) {
        synchronized(this) {
            fluxSinks.add(newFluxSinkObject);

razorcd

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

razorcd

@Repository
@RequiredArgsConstructor
public class CustomerRepository<E extends Identifiable> {
    
    private final RedisTemplate<String,E> staticTemplate;

    public void save(E event) {
      staticTemplate.opsForStream().add(ObjectRecord.create(customer.getStringId(), event));

    };

razorcd

@Component
@RequiredArgsConstructor
public class CustomerListener {

    private final CustomerRepository<CustomerUpdatedEvent> customerRepository;

    @StreamListener("customer-events")
    public void customerEventsConsumer(Flux<DataUpdatedEvent> event) {
        event.map(this::parseEvent)
             .doOnNext(MyService::doBusinessLogic)   // returns CustomerUpdatedEvent

razorcd

                                                                    +-------------- +
Kafka Listener  --> Business Logic --> Repository(RedisClient) -->  |               |
                                                                    | Redis Streams |
API Controller  -->  Repository(Redis Puller) ------------------->  |               |
                                                                    +---------------+

razorcd

Kafka Topic A, Partition 3 --> Service Instance 2 --> Reactive Mongo --> Service Instance 1 --> Event stream listener /customer1