Feature - takeFirstElements and takeLastElements stream methods added to RScoredSortedSetReactive object

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

@@ -113,16 +113,18 @@ public interface RBlockingDequeReactive<V> extends RDequeReactive<V>, RBlockingQ
     Mono<V> takeFirst();
 
     /**
-     * Retrieves and removes stream of elements from the head of this queue. Waits for an element become available.
+     * Retrieves and removes continues stream of elements from the head of this queue. 
+     * Waits for next element become available.
      * 
-     * @return the head element of this queue
+     * @return stream of head elements
      */
     Flux<V> takeFirstElements();
     
     /**
-     * Retrieves and removes stream of elements from the tail of this queue. Waits for an element become available.
+     * Retrieves and removes continues stream of elements from the tail of this queue. 
+     * Waits for next element become available.
      * 
-     * @return the head element of this queue
+     * @return stream of tail elements
      */
     Flux<V> takeLastElements();
     

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

@@ -156,9 +156,10 @@ public interface RBlockingQueueReactive<V> extends RQueueReactive<V> {
     Mono<Void> put(V e);
 
     /**
-     * Retrieves and removes stream of elements from the head of this queue.
+     * Retrieves and removes continues stream of elements from the head of this queue.
+     * Waits for next element become available.
      * 
-     * @return stream of messages
+     * @return stream of elements
      */
     Flux<V> takeElements();
     

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

@@ -431,4 +431,20 @@ public interface RScoredSortedSetReactive<V> extends RExpirableReactive, RSortab
      */
     Mono<V> takeLast();
 
+    /**
+     * Retrieves and removes continues stream of elements from the head of this queue. 
+     * Waits for next element become available.
+     * 
+     * @return stream of head elements
+     */
+    Flux<V> takeFirstElements();
+    
+    /**
+     * Retrieves and removes continues stream of elements from the tail of this queue. 
+     * Waits for next element become available.
+     * 
+     * @return stream of tail elements
+     */
+    Flux<V> takeLastElements();
+
 }

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

@@ -0,0 +1,77 @@
+/**
+ * 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.Callable;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.redisson.api.RFuture;
+
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.FutureListener;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
+
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
+public class ElementsStream {
+
+    private static <V> void take(final Callable<RFuture<V>> factory, final FluxSink<V> emitter, final AtomicLong counter, final AtomicReference<RFuture<V>> futureRef) {
+        RFuture<V> future;
+        try {
+            future = factory.call();
+        } catch (Exception e) {
+            emitter.error(e);
+            return;
+        }
+        futureRef.set(future);
+        future.addListener(new FutureListener<V>() {
+            @Override
+            public void operationComplete(Future<V> future) throws Exception {
+                if (!future.isSuccess()) {
+                    emitter.error(future.cause());
+                    return;
+                }
+                
+                emitter.next(future.getNow());
+                if (counter.decrementAndGet() == 0) {
+                    emitter.complete();
+                }
+                
+                take(factory, emitter, counter, futureRef);
+            }
+        });
+    }
+    
+    public static <V> Flux<V> takeElements(Callable<RFuture<V>> callable) {
+        return Flux.<V>create(emitter -> {
+            emitter.onRequest(n -> {
+                AtomicLong counter = new AtomicLong(n);
+                AtomicReference<RFuture<V>> futureRef = new AtomicReference<RFuture<V>>();
+                take(callable, emitter, counter, futureRef);
+                emitter.onDispose(() -> {
+                    futureRef.get().cancel(true);
+                });
+            });
+        });
+    }
+
+    
+}

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

@@ -38,7 +38,7 @@ public class RedissonBlockingDequeReactive<V> extends RedissonBlockingQueueReact
     }
 
     public Flux<V> takeFirstElements() {
-        return takeElements(new Callable<RFuture<V>>() {
+        return ElementsStream.takeElements(new Callable<RFuture<V>>() {
             @Override
             public RFuture<V> call() throws Exception {
                 return queue.takeFirstAsync();
@@ -47,7 +47,7 @@ public class RedissonBlockingDequeReactive<V> extends RedissonBlockingQueueReact
     }
     
     public Flux<V> takeLastElements() {
-        return takeElements(new Callable<RFuture<V>>() {
+        return ElementsStream.takeElements(new Callable<RFuture<V>>() {
             @Override
             public RFuture<V> call() throws Exception {
                 return queue.takeLastAsync();

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

@@ -16,17 +16,12 @@
 package org.redisson.reactive;
 
 import java.util.concurrent.Callable;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
 
 import org.redisson.api.RBlockingQueue;
 import org.redisson.api.RFuture;
 import org.redisson.api.RListAsync;
 
-import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.FutureListener;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.FluxSink;
 
 /**
  * 
@@ -43,35 +38,8 @@ public class RedissonBlockingQueueReactive<V> extends RedissonListReactive<V> {
         this.queue = queue;
     }
 
-    private void take(final Callable<RFuture<V>> factory, final FluxSink<V> emitter, final AtomicLong counter, final AtomicReference<RFuture<V>> futureRef) {
-        RFuture<V> future;
-        try {
-            future = factory.call();
-        } catch (Exception e) {
-            emitter.error(e);
-            return;
-        }
-        futureRef.set(future);
-        future.addListener(new FutureListener<V>() {
-            @Override
-            public void operationComplete(Future<V> future) throws Exception {
-                if (!future.isSuccess()) {
-                    emitter.error(future.cause());
-                    return;
-                }
-                
-                emitter.next(future.getNow());
-                if (counter.decrementAndGet() == 0) {
-                    emitter.complete();
-                }
-                
-                take(factory, emitter, counter, futureRef);
-            }
-        });
-    }
-    
     public Flux<V> takeElements() {
-        return takeElements(new Callable<RFuture<V>>() {
+        return ElementsStream.takeElements(new Callable<RFuture<V>>() {
             @Override
             public RFuture<V> call() throws Exception {
                 return queue.takeAsync();
@@ -79,17 +47,4 @@ public class RedissonBlockingQueueReactive<V> extends RedissonListReactive<V> {
         });
     }
     
-    protected final Flux<V> takeElements(Callable<RFuture<V>> callable) {
-        return Flux.<V>create(emitter -> {
-            emitter.onRequest(n -> {
-                AtomicLong counter = new AtomicLong(n);
-                AtomicReference<RFuture<V>> futureRef = new AtomicReference<RFuture<V>>();
-                take(callable, emitter, counter, futureRef);
-                emitter.onDispose(() -> {
-                    futureRef.get().cancel(true);
-                });
-            });
-        });
-    }
-    
 }

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

@@ -15,7 +15,8 @@
  */
 package org.redisson.reactive;
 
-import org.reactivestreams.Publisher;
+import java.util.concurrent.Callable;
+
 import org.redisson.RedissonScoredSortedSet;
 import org.redisson.api.RFuture;
 import org.redisson.api.RScoredSortedSetAsync;
@@ -51,6 +52,24 @@ public class RedissonScoredSortedSetReactive<V>  {
         this.instance = instance;
     }
     
+    public Flux<V> takeFirstElements() {
+        return ElementsStream.takeElements(new Callable<RFuture<V>>() {
+            @Override
+            public RFuture<V> call() throws Exception {
+                return instance.takeFirstAsync();
+            }
+        });
+    }
+    
+    public Flux<V> takeLastElements() {
+        return ElementsStream.takeElements(new Callable<RFuture<V>>() {
+            @Override
+            public RFuture<V> call() throws Exception {
+                return instance.takeLastAsync();
+            }
+        });
+    }
+
     private Flux<V> scanIteratorReactive(final String pattern, final int count) {
         return Flux.create(new SetReactiveIterator<V>() {
             @Override