public <M> Flowable<M> getMessages(Class<M> type) {
        ReplayProcessor<M> p = ReplayProcessor.create();
        return p.doOnRequest(new LongConsumer() {
            @Override
            public void accept(long n) throws Exception {
                AtomicLong counter = new AtomicLong(n);
                RFuture<Integer> t = topic.addListenerAsync(type, new MessageListener<M>() {
                    @Override
                    public void onMessage(CharSequence channel, M msg) {
                        p.onNext(msg);
                        if (counter.decrementAndGet() == 0) {
                            topic.removeListenerAsync(this);
                            p.onComplete();
                        }
                    }
                });
                t.whenComplete((id, e) -> {
                    if (e != null) {
                        p.onError(e);
                        return;
                    }
                    
                    p.doOnCancel(new Action() {
                        @Override
                        public void run() throws Exception {
                            topic.removeListenerAsync(id);
                        }
                    });
                });
            }
        });
    }
@Test
    public void testLong() throws InterruptedException {
        RTopicRx topic = redisson.getTopic("test");
        Flowable<String> messages = topic.getMessages(String.class);
        List<String> list = new ArrayList<>();
        messages.subscribe(new Subscriber<String>() {

            @Override
            public void onSubscribe(Subscription s) {
                s.request(10);
            }

            @Override
            public void onNext(String t) {
                list.add(t);
            }

            @Override
            public void onError(Throwable t) {
            }

            @Override
            public void onComplete() {
            }
        });
        
        for (int i = 0; i < 15; i++) {
            sync(topic.publish("" + i));
        }

        Awaitility.waitAtMost(Duration.ofSeconds(10)).until(() -> list.equals(Arrays.asList("0", "1", "2", "3", "4", "5", "6", "7", "8", "9")));
    }