fix cancel on terminal event with MultiSubscriptionSubscriber

reactor · Jun 10, 2016 · 36aa312 · 36aa312
1 parent bd8c148
commit 36aa312
Show file tree

Hide file tree

Showing 9 changed files with 47 additions and 3 deletions.
diff --git a/src/main/java/reactor/core/publisher/FluxRepeat.java b/src/main/java/reactor/core/publisher/FluxRepeat.java
@@ -95,6 +95,11 @@ public void onNext(T t) {
 			subscriber.onNext(t);
 		}
 
+		@Override
+		protected boolean shouldCancelCurrent() {
+			return false;
+		}
+
 		@Override
 		public void onComplete() {
 			long r = remaining;

diff --git a/src/main/java/reactor/core/publisher/FluxRepeatPredicate.java b/src/main/java/reactor/core/publisher/FluxRepeatPredicate.java
@@ -82,6 +82,11 @@ public RepeatPredicateSubscriber(Publisher<? extends T> source,
 			this.predicate = predicate;
 		}
 
+		@Override
+		protected boolean shouldCancelCurrent() {
+			return false;
+		}
+
 		@Override
 		public void onNext(T t) {
 			produced++;

diff --git a/src/main/java/reactor/core/publisher/FluxRepeatWhen.java b/src/main/java/reactor/core/publisher/FluxRepeatWhen.java
@@ -121,6 +121,11 @@ public RepeatWhenMainSubscriber(Subscriber<? super T> actual, Subscriber<Long> s
 			this.otherArbiter = new DeferredSubscription();
 		}
 
+		@Override
+		protected boolean shouldCancelCurrent() {
+			return false;
+		}
+
 		@Override
 		public void cancel() {
 			if (cancelled) {

diff --git a/src/main/java/reactor/core/publisher/FluxResume.java b/src/main/java/reactor/core/publisher/FluxResume.java
@@ -69,6 +69,11 @@ public ResumeSubscriber(Subscriber<? super T> actual,
 			this.nextFactory = nextFactory;
 		}
 
+		@Override
+		protected boolean shouldCancelCurrent() {
+			return false;
+		}
+
 		@Override
 		public void onSubscribe(Subscription s) {
 			if (!second) {

diff --git a/src/main/java/reactor/core/publisher/FluxRetry.java b/src/main/java/reactor/core/publisher/FluxRetry.java
@@ -103,6 +103,11 @@ public void onError(Throwable t) {
 			resubscribe();
 		}
 
+		@Override
+		protected boolean shouldCancelCurrent() {
+			return false;
+		}
+
 		void resubscribe() {
 			if (WIP.getAndIncrement(this) == 0) {
 				do {

diff --git a/src/main/java/reactor/core/publisher/FluxRetryPredicate.java b/src/main/java/reactor/core/publisher/FluxRetryPredicate.java
@@ -89,6 +89,11 @@ public void onNext(T t) {
 			subscriber.onNext(t);
 		}
 
+		@Override
+		protected boolean shouldCancelCurrent() {
+			return false;
+		}
+
 		@Override
 		public void onError(Throwable t) {
 			boolean b;

diff --git a/src/main/java/reactor/core/publisher/FluxRetryWhen.java b/src/main/java/reactor/core/publisher/FluxRetryWhen.java
@@ -202,6 +202,11 @@ void whenComplete() {
 
 			subscriber.onComplete();
 		}
+
+		@Override
+		protected boolean shouldCancelCurrent() {
+			return false;
+		}
 	}
 
 	static final class RetryWhenOtherSubscriber

diff --git a/src/main/java/reactor/core/publisher/FluxSwitchIfEmpty.java b/src/main/java/reactor/core/publisher/FluxSwitchIfEmpty.java
@@ -85,6 +85,11 @@ public void onComplete() {
 			}
 		}
 
+		@Override
+		protected boolean shouldCancelCurrent() {
+			return false;
+		}
+
 		@Override
 		public Object connectedInput() {
 			return null;

diff --git a/src/main/java/reactor/core/subscriber/MultiSubscriptionSubscriber.java b/src/main/java/reactor/core/subscriber/MultiSubscriptionSubscriber.java
@@ -112,8 +112,8 @@ public final void set(Subscription s) {
 
 		if (wip == 0 && WIP.compareAndSet(this, 0, 1)) {
 			Subscription a = actual;
-			
-			if (a != null) {
+
+			if (a != null && shouldCancelCurrent()) {
 				a.cancel();
 			}
 
@@ -311,7 +311,7 @@ final void drainLoop() {
 				}
 
 				if (ms != null) {
-					if (a != null) {
+					if (a != null && shouldCancelCurrent()) {
 						a.cancel();
 					}
 					actual = ms;
@@ -358,4 +358,8 @@ public boolean isStarted() {
 	public final boolean isUnbounded() {
 		return unbounded;
 	}
+
+	protected boolean shouldCancelCurrent() {
+		return true;
+	}
 }