Feature - RTopicReactive.getMessages streaming method added

redisson/src/main/java/org/redisson/RedissonReactive.java

@@ -80,6 +80,7 @@ import org.redisson.reactive.RedissonScoredSortedSetReactive;
 import org.redisson.reactive.RedissonSetCacheReactive;
 import org.redisson.reactive.RedissonSetMultimapReactive;
 import org.redisson.reactive.RedissonSetReactive;
+import org.redisson.reactive.RedissonTopicReactive;
 import org.redisson.reactive.RedissonTransactionReactive;
 
 /**
@@ -295,12 +296,16 @@ public class RedissonReactive implements RedissonReactiveClient {
 
     @Override
     public RTopicReactive getTopic(String name) {
-        return ReactiveProxyBuilder.create(commandExecutor, new RedissonTopic(commandExecutor, name), RTopicReactive.class);
+        RedissonTopic topic = new RedissonTopic(commandExecutor, name);
+        return ReactiveProxyBuilder.create(commandExecutor, topic,
+                new RedissonTopicReactive(topic), RTopicReactive.class);
     }
 
     @Override
     public RTopicReactive getTopic(String name, Codec codec) {
-        return ReactiveProxyBuilder.create(commandExecutor, new RedissonTopic(codec, commandExecutor, name), RTopicReactive.class);
+        RedissonTopic topic = new RedissonTopic(codec, commandExecutor, name);
+        return ReactiveProxyBuilder.create(commandExecutor, topic, 
+                new RedissonTopicReactive(topic), RTopicReactive.class);
     }
 
     @Override

redisson/src/main/java/org/redisson/api/RTopicReactive.java

@@ -20,6 +20,7 @@ import java.util.List;
 import org.redisson.api.listener.MessageListener;
 import org.redisson.api.listener.StatusListener;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 /**
@@ -71,6 +72,15 @@ public interface RTopicReactive {
      * Removes the listener by <code>id</code> for listening this topic
      *
      * @param listenerId - listener id
+     * @return void
      */
-    void removeListener(int listenerId);
+    Mono<Void> removeListener(int listenerId);
+    
+    /**
+     * Returns stream of messages.
+     * 
+     * @param type - type of message to listen
+     * @return stream of messages
+     */
+    <M> Flux<M> getMessages(Class<M> type);
 }

redisson/src/main/java/org/redisson/reactive/RedissonTopicReactive.java

@@ -0,0 +1,62 @@
+/**
+ * Copyright 2018 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.reactive;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.redisson.api.RFuture;
+import org.redisson.api.RTopic;
+import org.redisson.api.listener.MessageListener;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
+public class RedissonTopicReactive {
+
+    private final RTopic topic;
+    
+    public RedissonTopicReactive(RTopic topic) {
+        this.topic = topic;
+    }
+
+    public <M> Flux<M> getMessages(Class<M> type) {
+        return Flux.<M>create(emitter -> {
+            emitter.onRequest(n -> {
+                AtomicLong counter = new AtomicLong(n);
+                RFuture<Integer> t = topic.addListenerAsync(type, new MessageListener<M>() {
+                    @Override
+                    public void onMessage(CharSequence channel, M msg) {
+                        emitter.next(msg);
+                        if (counter.decrementAndGet() == 0) {
+                            topic.removeListenerAsync(this);
+                            emitter.complete();
+                        }
+                    }
+                });
+                t.whenComplete((id, e) -> {
+                    emitter.onDispose(() -> {
+                        topic.removeListenerAsync(id);
+                    });
+                });
+            });
+        });
+    }
+    
+}

redisson/src/test/java/org/redisson/RedissonTopicReactiveTest.java

@@ -0,0 +1,49 @@
+package org.redisson;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.redisson.api.RTopicReactive;
+
+import reactor.core.publisher.Flux;
+
+public class RedissonTopicReactiveTest extends BaseReactiveTest {
+
+    @Test
+    public void testLong() throws InterruptedException {
+        RTopicReactive topic = redisson.getTopic("test");
+        Flux<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));
+        }
+        
+        assertThat(list).containsExactly("0", "1", "2", "3", "4", "5", "6", "7", "8", "9");
+    }
+}