Scaling reactive APIs - www.razorcodes.com

code7_RedisPuller2.java

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

    @Override
    public void close() {
        this.pullerThread.interrupt();
    }

    private Thread getPullerThread(Map<String,EventStream> streamsList) {
        return new Thread(() -> {
            while (true) {
                try {
                    staticRedisPoller(streamsList);
                } catch (Exception e) {
                    log.info("Closing RedisPuller#{}: Error: {}", Thread.currentThread(), e);
                } finally {
                    Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS); // retry time
                }
            }
        });
    }

    //... puller methods like above
}