diff --git a/redisson/src/main/java/org/redisson/api/RBlockingDequeReactive.java b/redisson/src/main/java/org/redisson/api/RBlockingDequeReactive.java index deaaa48d8..f72ddb7c1 100644 --- a/redisson/src/main/java/org/redisson/api/RBlockingDequeReactive.java +++ b/redisson/src/main/java/org/redisson/api/RBlockingDequeReactive.java @@ -113,16 +113,18 @@ public interface RBlockingDequeReactive extends RDequeReactive, RBlockingQ Mono 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 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 takeLastElements(); diff --git a/redisson/src/main/java/org/redisson/api/RBlockingQueueReactive.java b/redisson/src/main/java/org/redisson/api/RBlockingQueueReactive.java index d81796985..13dea7281 100644 --- a/redisson/src/main/java/org/redisson/api/RBlockingQueueReactive.java +++ b/redisson/src/main/java/org/redisson/api/RBlockingQueueReactive.java @@ -156,9 +156,10 @@ public interface RBlockingQueueReactive extends RQueueReactive { Mono 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 takeElements(); diff --git a/redisson/src/main/java/org/redisson/api/RScoredSortedSetReactive.java b/redisson/src/main/java/org/redisson/api/RScoredSortedSetReactive.java index e5e50c536..7c46c6cff 100644 --- a/redisson/src/main/java/org/redisson/api/RScoredSortedSetReactive.java +++ b/redisson/src/main/java/org/redisson/api/RScoredSortedSetReactive.java @@ -430,5 +430,21 @@ public interface RScoredSortedSetReactive extends RExpirableReactive, RSortab * @return the tail element */ Mono 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 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 takeLastElements(); + } diff --git a/redisson/src/main/java/org/redisson/reactive/ElementsStream.java b/redisson/src/main/java/org/redisson/reactive/ElementsStream.java new file mode 100644 index 000000000..a44b08f43 --- /dev/null +++ b/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 void take(final Callable> factory, final FluxSink emitter, final AtomicLong counter, final AtomicReference> futureRef) { + RFuture future; + try { + future = factory.call(); + } catch (Exception e) { + emitter.error(e); + return; + } + futureRef.set(future); + future.addListener(new FutureListener() { + @Override + public void operationComplete(Future 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 Flux takeElements(Callable> callable) { + return Flux.create(emitter -> { + emitter.onRequest(n -> { + AtomicLong counter = new AtomicLong(n); + AtomicReference> futureRef = new AtomicReference>(); + take(callable, emitter, counter, futureRef); + emitter.onDispose(() -> { + futureRef.get().cancel(true); + }); + }); + }); + } + + +} diff --git a/redisson/src/main/java/org/redisson/reactive/RedissonBlockingDequeReactive.java b/redisson/src/main/java/org/redisson/reactive/RedissonBlockingDequeReactive.java index 0f14e07ee..bb9d46dc2 100644 --- a/redisson/src/main/java/org/redisson/reactive/RedissonBlockingDequeReactive.java +++ b/redisson/src/main/java/org/redisson/reactive/RedissonBlockingDequeReactive.java @@ -38,7 +38,7 @@ public class RedissonBlockingDequeReactive extends RedissonBlockingQueueReact } public Flux takeFirstElements() { - return takeElements(new Callable>() { + return ElementsStream.takeElements(new Callable>() { @Override public RFuture call() throws Exception { return queue.takeFirstAsync(); @@ -47,7 +47,7 @@ public class RedissonBlockingDequeReactive extends RedissonBlockingQueueReact } public Flux takeLastElements() { - return takeElements(new Callable>() { + return ElementsStream.takeElements(new Callable>() { @Override public RFuture call() throws Exception { return queue.takeLastAsync(); diff --git a/redisson/src/main/java/org/redisson/reactive/RedissonBlockingQueueReactive.java b/redisson/src/main/java/org/redisson/reactive/RedissonBlockingQueueReactive.java index a40c26e52..e5f73057b 100644 --- a/redisson/src/main/java/org/redisson/reactive/RedissonBlockingQueueReactive.java +++ b/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 extends RedissonListReactive { this.queue = queue; } - private void take(final Callable> factory, final FluxSink emitter, final AtomicLong counter, final AtomicReference> futureRef) { - RFuture future; - try { - future = factory.call(); - } catch (Exception e) { - emitter.error(e); - return; - } - futureRef.set(future); - future.addListener(new FutureListener() { - @Override - public void operationComplete(Future 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 takeElements() { - return takeElements(new Callable>() { + return ElementsStream.takeElements(new Callable>() { @Override public RFuture call() throws Exception { return queue.takeAsync(); @@ -79,17 +47,4 @@ public class RedissonBlockingQueueReactive extends RedissonListReactive { }); } - protected final Flux takeElements(Callable> callable) { - return Flux.create(emitter -> { - emitter.onRequest(n -> { - AtomicLong counter = new AtomicLong(n); - AtomicReference> futureRef = new AtomicReference>(); - take(callable, emitter, counter, futureRef); - emitter.onDispose(() -> { - futureRef.get().cancel(true); - }); - }); - }); - } - } diff --git a/redisson/src/main/java/org/redisson/reactive/RedissonScoredSortedSetReactive.java b/redisson/src/main/java/org/redisson/reactive/RedissonScoredSortedSetReactive.java index 21afb5506..ff92b4ebd 100644 --- a/redisson/src/main/java/org/redisson/reactive/RedissonScoredSortedSetReactive.java +++ b/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; @@ -50,6 +51,24 @@ public class RedissonScoredSortedSetReactive { private RedissonScoredSortedSetReactive(Codec codec, CommandReactiveExecutor commandExecutor, String name, RScoredSortedSetAsync instance) { this.instance = instance; } + + public Flux takeFirstElements() { + return ElementsStream.takeElements(new Callable>() { + @Override + public RFuture call() throws Exception { + return instance.takeFirstAsync(); + } + }); + } + + public Flux takeLastElements() { + return ElementsStream.takeElements(new Callable>() { + @Override + public RFuture call() throws Exception { + return instance.takeLastAsync(); + } + }); + } private Flux scanIteratorReactive(final String pattern, final int count) { return Flux.create(new SetReactiveIterator() {