tweaks MultiSubscription cancel

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

Hide file tree

Showing 11 changed files with 6 additions and 51 deletions.
diff --git a/src/main/java/reactor/core/publisher/FluxConcatMap.java b/src/main/java/reactor/core/publisher/FluxConcatMap.java
@@ -760,11 +760,6 @@ public ConcatMapInner(StreamConcatMapSupport<R> parent) {
 			this.parent = parent;
 		}
 
-		@Override
-		public void onSubscribe(Subscription s) {
-			set(s);
-		}
-
 		@Override
 		public void onNext(R t) {
 			produced++;

diff --git a/src/main/java/reactor/core/publisher/FluxRepeat.java b/src/main/java/reactor/core/publisher/FluxRepeat.java
@@ -95,11 +95,6 @@ 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,11 +82,6 @@ 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,11 +121,6 @@ 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) {
@@ -138,11 +133,6 @@ public void cancel() {
 			super.cancel();
 		}
 
-		@Override
-		public void onSubscribe(Subscription s) {
-			set(s);
-		}
-
 		@Override
 		public void onNext(T t) {
 			subscriber.onNext(t);

diff --git a/src/main/java/reactor/core/publisher/FluxResume.java b/src/main/java/reactor/core/publisher/FluxResume.java
@@ -69,11 +69,6 @@ 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,11 +103,6 @@ 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,11 +89,6 @@ 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,11 +202,6 @@ 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,11 +85,6 @@ public void onComplete() {
 			}
 		}
 
-		@Override
-		protected boolean shouldCancelCurrent() {
-			return false;
-		}
-
 		@Override
 		public Object connectedInput() {
 			return null;

diff --git a/src/main/java/reactor/core/publisher/FluxTimeout.java b/src/main/java/reactor/core/publisher/FluxTimeout.java
@@ -124,6 +124,11 @@ public void onSubscribe(Subscription s) {
 			}
 		}
 
+		@Override
+		protected boolean shouldCancelCurrent() {
+			return true;
+		}
+
 		@Override
 		public void onNext(T t) {
 			timeout.cancel();

diff --git a/src/main/java/reactor/core/subscriber/MultiSubscriptionSubscriber.java b/src/main/java/reactor/core/subscriber/MultiSubscriptionSubscriber.java
@@ -360,6 +360,6 @@ public final boolean isUnbounded() {
 	}
 
 	protected boolean shouldCancelCurrent() {
-		return true;
+		return false;
 	}
 }