Fixed - RCollectionReactive.addAll is executed without subscription. #4554

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

@@ -15,16 +15,14 @@
  */
 package org.redisson.reactive;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicLong;
-
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
 import org.redisson.api.RFuture;
-
 import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Mono;
 
+import java.util.concurrent.atomic.AtomicLong;
+
 /**
  * 
  * @author Nikita Koksharov
@@ -36,49 +34,48 @@ public abstract class PublisherAdder<V> {
     public abstract RFuture<Boolean> add(Object o);
 
     public Publisher<Boolean> addAll(Publisher<? extends V> c) {
-        CompletableFuture<Boolean> promise = new CompletableFuture<>();
-        c.subscribe(new BaseSubscriber<V>() {
-
-            volatile boolean completed;
-            AtomicLong values = new AtomicLong();
-            Subscription s;
-            Boolean lastSize = false;
+        return Mono.create(emitter -> emitter.onRequest(n -> {
+            c.subscribe(new BaseSubscriber<V>() {
 
-            @Override
-            protected void hookOnSubscribe(Subscription s) {
-                this.s = s;
-                s.request(1);
-            }
+                volatile boolean completed;
+                final AtomicLong values = new AtomicLong();
+                Subscription s;
+                volatile Boolean lastSize = false;
 
-            @Override
-            protected void hookOnNext(V o) {
-                values.getAndIncrement();
-                add(o).whenComplete((res, e) -> {
-                    if (e != null) {
-                        promise.completeExceptionally(e);
-                        return;
-                    }
-                    
-                    if (res) {
-                        lastSize = true;
-                    }
+                @Override
+                protected void hookOnSubscribe(Subscription s) {
+                    this.s = s;
                     s.request(1);
-                    if (values.decrementAndGet() == 0 && completed) {
-                        promise.complete(lastSize);
-                    }
-                });
-            }
+                }
 
-            @Override
-            protected void hookOnComplete() {
-                completed = true;
-                if (values.get() == 0) {
-                    promise.complete(lastSize);
+                @Override
+                protected void hookOnNext(V o) {
+                    values.getAndIncrement();
+                    add(o).whenComplete((res, e) -> {
+                        if (e != null) {
+                            emitter.error(e);
+                            return;
+                        }
+
+                        if (res) {
+                            lastSize = true;
+                        }
+                        s.request(1);
+                        if (values.decrementAndGet() == 0 && completed) {
+                            emitter.success(lastSize);
+                        }
+                    });
                 }
-            }
-        });
 
-        return Mono.fromCompletionStage(promise);
+                @Override
+                protected void hookOnComplete() {
+                    completed = true;
+                    if (values.get() == 0) {
+                        emitter.success(lastSize);
+                    }
+                }
+            });
+        }));
     }
 
 }

redisson/src/test/java/org/redisson/RedissonListReactiveTest.java

@@ -3,9 +3,12 @@ package org.redisson;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.redisson.api.RListReactive;
+import org.redisson.api.RedissonReactiveClient;
 import org.redisson.client.RedisException;
+import org.redisson.client.codec.StringCodec;
 import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import java.util.Arrays;
 import java.util.Collections;
@@ -18,6 +21,16 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 public class RedissonListReactiveTest extends BaseReactiveTest {
 
+    @Test
+    public void test1() throws InterruptedException {
+        RListReactive<String> testQueue = redisson.getList("list");
+        Mono<Boolean> s = testQueue.addAll(Flux.just("a", "b", "c"));
+        Thread.sleep(400);
+        assertThat(testQueue.iterator().collectList().block()).isEmpty();
+        assertThat(s.block()).isTrue();
+        assertThat(testQueue.iterator().collectList().block()).containsExactly("a", "b", "c");
+    }
+
     @Test
     public void testIteratorFilter() {
         for (int i = 0; i < 10; i++) {