fix reactor#389 - support conditional subscriber in parallelflux (rea…

…ctor#865)
elvinshang · Sep 19, 2017 · c3c1e6e · c3c1e6e
1 parent f7a2e00
commit c3c1e6e
Show file tree

Hide file tree

Showing 10 changed files with 154 additions and 26 deletions.
diff --git a/reactor-core/src/main/java/reactor/core/publisher/ParallelFilter.java b/reactor-core/src/main/java/reactor/core/publisher/ParallelFilter.java
@@ -19,6 +19,7 @@
 import javax.annotation.Nullable;
 
 import reactor.core.CoreSubscriber;
+import reactor.core.Fuseable;
 import reactor.core.Scannable;
 
 /**
@@ -47,19 +48,28 @@ public Object scanUnsafe(Attr key) {
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public void subscribe(CoreSubscriber<? super T>[] subscribers) {
 		if (!validate(subscribers)) {
 			return;
 		}
 
 		int n = subscribers.length;
-		@SuppressWarnings("unchecked")
+
 		CoreSubscriber<? super T>[] parents = new CoreSubscriber[n];
-
+
+		boolean conditional = subscribers[0] instanceof Fuseable.ConditionalSubscriber;
+
 		for (int i = 0; i < n; i++) {
-			parents[i] = new FluxFilter.FilterSubscriber<>(subscribers[i], predicate);
+			if (conditional) {
+				parents[i] = new FluxFilter.FilterConditionalSubscriber<>(
+						(Fuseable.ConditionalSubscriber<T>)subscribers[i], predicate);
+			}
+			else {
+				parents[i] = new FluxFilter.FilterSubscriber<>(subscribers[i], predicate);
+			}
 		}
-		
+
 		source.subscribe(parents);
 	}
 

diff --git a/reactor-core/src/main/java/reactor/core/publisher/ParallelLog.java b/reactor-core/src/main/java/reactor/core/publisher/ParallelLog.java
@@ -18,6 +18,7 @@
 import javax.annotation.Nullable;
 
 import reactor.core.CoreSubscriber;
+import reactor.core.Fuseable;
 import reactor.core.Scannable;
 
 /**
@@ -39,17 +40,25 @@ final class ParallelLog<T> extends ParallelFlux<T> implements Scannable {
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public void subscribe(CoreSubscriber<? super T>[] subscribers) {
 		if (!validate(subscribers)) {
 			return;
 		}
 
 		int n = subscribers.length;
-		@SuppressWarnings("unchecked")
 		CoreSubscriber<? super T>[] parents = new CoreSubscriber[n];
-
+
+		boolean conditional = subscribers[0] instanceof Fuseable.ConditionalSubscriber;
+
 		for (int i = 0; i < n; i++) {
-			parents[i] = new FluxPeek.PeekSubscriber<>(subscribers[i], log);
+			if (conditional) {
+				parents[i] = new FluxPeekFuseable.PeekConditionalSubscriber<>(
+						(Fuseable.ConditionalSubscriber<T>)subscribers[i], log);
+			}
+			else {
+				parents[i] = new FluxPeek.PeekSubscriber<>(subscribers[i], log);
+			}
 		}
 
 		source.subscribe(parents);

diff --git a/reactor-core/src/main/java/reactor/core/publisher/ParallelMap.java b/reactor-core/src/main/java/reactor/core/publisher/ParallelMap.java
@@ -19,6 +19,7 @@
 import javax.annotation.Nullable;
 
 import reactor.core.CoreSubscriber;
+import reactor.core.Fuseable;
 import reactor.core.Scannable;
 
 /**
@@ -48,17 +49,26 @@ public Object scanUnsafe(Attr key) {
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public void subscribe(CoreSubscriber<? super R>[] subscribers) {
 		if (!validate(subscribers)) {
 			return;
 		}
 
 		int n = subscribers.length;
-		@SuppressWarnings("unchecked")
 		CoreSubscriber<? super T>[] parents = new CoreSubscriber[n];
-
+
+		boolean conditional = subscribers[0] instanceof Fuseable.ConditionalSubscriber;
+
 		for (int i = 0; i < n; i++) {
-			parents[i] = new FluxMap.MapSubscriber<>(subscribers[i], mapper);
+			if (conditional) {
+				parents[i] =
+						new FluxMap.MapConditionalSubscriber<>((Fuseable.ConditionalSubscriber<R>) subscribers[i],
+								mapper);
+			}
+			else {
+				parents[i] = new FluxMap.MapSubscriber<>(subscribers[i], mapper);
+			}
 		}
 
 		source.subscribe(parents);

diff --git a/reactor-core/src/main/java/reactor/core/publisher/ParallelPeek.java b/reactor-core/src/main/java/reactor/core/publisher/ParallelPeek.java
@@ -21,6 +21,7 @@
 
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
+import reactor.core.Fuseable;
 
 /**
  * Execute a Consumer in each 'rail' for the current element passing through.
@@ -63,17 +64,26 @@ final class ParallelPeek<T> extends ParallelFlux<T> implements SignalPeek<T>{
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public void subscribe(CoreSubscriber<? super T>[] subscribers) {
 		if (!validate(subscribers)) {
 			return;
 		}
 
 		int n = subscribers.length;
-		@SuppressWarnings("unchecked")
+
 		CoreSubscriber<? super T>[] parents = new CoreSubscriber[n];
-
+
+		boolean conditional = subscribers[0] instanceof Fuseable.ConditionalSubscriber;
+
 		for (int i = 0; i < n; i++) {
-			parents[i] = new FluxPeek.PeekSubscriber<>(subscribers[i], this);
+			if (conditional) {
+				parents[i] = new FluxPeekFuseable.PeekConditionalSubscriber<>(
+						(Fuseable.ConditionalSubscriber<T>)subscribers[i], this);
+			}
+			else {
+				parents[i] = new FluxPeek.PeekSubscriber<>(subscribers[i], this);
+			}
 		}
 
 		source.subscribe(parents);

diff --git a/reactor-core/src/main/java/reactor/core/publisher/ParallelRunOn.java b/reactor-core/src/main/java/reactor/core/publisher/ParallelRunOn.java
@@ -60,25 +60,33 @@ public Object scanUnsafe(Attr key) {
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public void subscribe(CoreSubscriber<? super T>[] subscribers) {
 		if (!validate(subscribers)) {
 			return;
 		}
 
 		int n = subscribers.length;
 
-		@SuppressWarnings("unchecked")
+
 		CoreSubscriber<T>[] parents = new CoreSubscriber[n];
-
+
+		boolean conditional = subscribers[0] instanceof Fuseable.ConditionalSubscriber;
+
 		for (int i = 0; i < n; i++) {
-			CoreSubscriber<? super T> a = subscribers[i];
-
 			Worker w = scheduler.createWorker();
 
-			CoreSubscriber<T> parent = new FluxPublishOn.PublishOnSubscriber<>(a,
-					scheduler, w, true,
-					prefetch, queueSupplier);
-			parents[i] = parent;
+			if (conditional) {
+				parents[i] = new FluxPublishOn.PublishOnConditionalSubscriber<>(
+						(Fuseable.ConditionalSubscriber<T>)subscribers[i],
+						scheduler, w, true,
+						prefetch, queueSupplier);
+			}
+			else {
+				parents[i] = new FluxPublishOn.PublishOnSubscriber<>(subscribers[i],
+						scheduler, w, true,
+						prefetch, queueSupplier);
+			}
 		}
 
 		source.subscribe(parents);

diff --git a/reactor-core/src/test/java/reactor/core/publisher/ParallelFilterTest.java b/reactor-core/src/test/java/reactor/core/publisher/ParallelFilterTest.java
@@ -5,7 +5,7 @@
  * 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
+ *       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,
@@ -18,6 +18,8 @@
 
 import org.junit.Test;
 import reactor.core.Scannable;
+import reactor.core.scheduler.Schedulers;
+import reactor.test.StepVerifier;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -44,4 +46,19 @@ public void scanOperator() throws Exception {
 				.isNotEqualTo(source.getPrefetch());
 	}
 
+	@Test
+	public void conditional() {
+		Flux<Integer> source = Flux.range(1, 1_000);
+		for (int i = 1; i < 33; i++) {
+			Flux<Integer> result = ParallelFlux.from(source, i)
+			                                   .filter(t -> true)
+			                                   .filter(t -> true)
+			                                   .sequential();
+
+			StepVerifier.create(result)
+			            .expectNextCount(1_000)
+			            .verifyComplete();
+		}
+	}
+
 }
diff --git a/reactor-core/src/test/java/reactor/core/publisher/ParallelLogTest.java b/reactor-core/src/test/java/reactor/core/publisher/ParallelLogTest.java
@@ -5,7 +5,7 @@
  * 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
+ *       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,
@@ -18,6 +18,8 @@
 
 import org.junit.Test;
 import reactor.core.Scannable;
+import reactor.core.scheduler.Schedulers;
+import reactor.test.StepVerifier;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -44,4 +46,18 @@ public void scanOperator() throws Exception {
 				.isEqualTo(source.getPrefetch());
 	}
 
+	@Test
+	public void conditional() {
+		Flux<Integer> source = Flux.range(1, 1_000);
+		for (int i = 1; i < 33; i++) {
+			Flux<Integer> result = ParallelFlux.from(source, i)
+			                                   .log()
+			                                   .filter(t -> true)
+			                                   .sequential();
+
+			StepVerifier.create(result)
+			            .expectNextCount(1_000)
+			            .verifyComplete();
+		}
+	}
 }
diff --git a/reactor-core/src/test/java/reactor/core/publisher/ParallelMapTest.java b/reactor-core/src/test/java/reactor/core/publisher/ParallelMapTest.java
@@ -5,7 +5,7 @@
  * 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
+ *       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,
@@ -18,6 +18,8 @@
 
 import org.junit.Test;
 import reactor.core.Scannable;
+import reactor.core.scheduler.Schedulers;
+import reactor.test.StepVerifier;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -41,4 +43,18 @@ public void scanOperator() {
 		assertThat(test.scan(Scannable.Attr.PREFETCH)).isEqualTo(-1);
 	}
 
+	@Test
+	public void conditional() {
+		Flux<Integer> source = Flux.range(1, 1_000);
+		for (int i = 1; i < 33; i++) {
+			Flux<Integer> result = ParallelFlux.from(source, i)
+			                                   .map(v -> v + 1)
+			                                   .filter(t -> true)
+			                                   .sequential();
+
+			StepVerifier.create(result)
+			            .expectNextCount(1_000)
+			            .verifyComplete();
+		}
+	}
 }
diff --git a/reactor-core/src/test/java/reactor/core/publisher/ParallelPeekTest.java b/reactor-core/src/test/java/reactor/core/publisher/ParallelPeekTest.java
@@ -5,7 +5,7 @@
  * 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
+ *       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,
@@ -18,6 +18,8 @@
 
 import org.junit.Test;
 import reactor.core.Scannable;
+import reactor.core.scheduler.Schedulers;
+import reactor.test.StepVerifier;
 import reactor.util.concurrent.Queues;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -45,4 +47,19 @@ public void scanOperator() {
 				.isEqualTo(test.getPrefetch())
 				.isEqualTo(123);
 	}
+
+	@Test
+	public void conditional() {
+		Flux<Integer> source = Flux.range(1, 1_000);
+		for (int i = 1; i < 33; i++) {
+			Flux<Integer> result = ParallelFlux.from(source, i)
+			                                   .doOnNext(d -> {})
+			                                   .filter(t -> true)
+			                                   .sequential();
+
+			StepVerifier.create(result)
+			            .expectNextCount(1_000)
+			            .verifyComplete();
+		}
+	}
 }
diff --git a/reactor-core/src/test/java/reactor/core/publisher/ParallelRunOnTest.java b/reactor-core/src/test/java/reactor/core/publisher/ParallelRunOnTest.java
@@ -5,7 +5,7 @@
  * 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
+ *       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,
@@ -19,6 +19,7 @@
 import org.junit.Test;
 import reactor.core.Scannable;
 import reactor.core.scheduler.Schedulers;
+import reactor.test.StepVerifier;
 import reactor.util.concurrent.Queues;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -42,4 +43,18 @@ public void parallelism() {
 		assertThat(test.parallelism()).isEqualTo(2);
 	}
 
+	@Test
+	public void conditional() {
+		Flux<Integer> source = Flux.range(1, 1_000);
+		for (int i = 1; i < 33; i++) {
+			Flux<Integer> result = ParallelFlux.from(source, i)
+			                                   .runOn(Schedulers.parallel())
+			                                   .filter(t -> true)
+			                                   .sequential();
+
+			StepVerifier.create(result)
+			            .expectNextCount(1_000)
+			            .verifyComplete();
+		}
+	}
 }