Fixing a race in pipeOn and tidying pipeOn tests (#40)

SoftwareAG · Sep 25, 2018 · 90d3861 · 90d3861
1 parent 5560197
commit 90d3861
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 22 deletions.
diff --git a/src/rx/operators/internals/PipeOn.mon b/src/rx/operators/internals/PipeOn.mon
@@ -63,6 +63,7 @@ event PipeOnOnConnection {
 
 		if not spawned and not spawning {
 			spawning := true;
+			monitor.subscribe(downstreamChannel); // Need to subscribe before spawning otherwise the spawned context could send Spawned() before we have subscribed
 			spawnToContext(downstreamChannel);
 		}
 		if spawning {
@@ -95,7 +96,7 @@ event PipeOnOnConnection {
 			ifpresent upstream {
 				upstream.dispose();
 			}
-			upstream := new IDisposable;
+			upstream := new optional<IDisposable>;
 
 			if spawned {
 				send Dispose(downstreamChannel) to downstreamChannel;

diff --git a/test/tests/Operators/PipeOn/Async/Input/test.mon b/test/tests/Operators/PipeOn/Async/Input/test.mon
@@ -24,16 +24,6 @@ monitor TestObservable {
 	      	.subscribeOn(Subscriber.create().onNext(checkContextNext).onComplete(checkContextComplete), testContext);
 	}
 
-	action generator(IResolver resolver) {
-		checkContext();
-		on all wait(0.1) and not wait(0.5) {
-			resolver.next("Value");
-		}
-		on wait(0.5) {
-			resolver.complete();
-		}
-	}
-
 	action checkContextNext(any value) {
 		checkContext();
 	}

diff --git a/test/tests/Operators/PipeOn/Sync/Input/test.mon b/test/tests/Operators/PipeOn/Sync/Input/test.mon
@@ -23,17 +23,7 @@ monitor TestObservable {
 			.do(ExpectValues.create([<any>0.0, 1.0, 2.0, 3.0], checkContext, test.fail))
 	      	.subscribeOn(Subscriber.create().onNext(checkContextNext).onComplete(checkContextComplete), testContext);
 	}
-
-	action generator(IResolver resolver) {
-		checkContext();
-		on all wait(0.1) and not wait(0.5) {
-			resolver.next("Value");
-		}
-		on wait(0.5) {
-			resolver.complete();
-		}
-	}
-
+
 	action checkContextNext(any value) {
 		checkContext();
 	}