Update .cancellable implementation to fix #946 (#949)

* Subscribe to upstream publisher instead of PassthroughSubject

* Fixed operator order

* Replaced RunLoop.immediate with RunLoop.test

Author: iampatbrown

Examples/VoiceMemos/VoiceMemosTests/VoiceMemosTests.swift

@@ -113,7 +113,7 @@ class VoiceMemosTests: XCTestCase {
     environment.audioRecorder.startRecording = { _ in
       audioRecorderSubject.eraseToEffect()
     }
-    environment.mainRunLoop = .immediate
+    environment.mainRunLoop = self.mainRunLoop.eraseToAnyScheduler()
     environment.temporaryDirectory = { .init(fileURLWithPath: "/tmp") }
     environment.uuid = { UUID(uuidString: "DEADBEEF-DEAD-BEEF-DEAD-BEEFDEADBEEF")! }
 
@@ -124,6 +124,7 @@ class VoiceMemosTests: XCTestCase {
     )
 
     store.send(.recordButtonTapped)
+    self.mainRunLoop.advance(by: 0.5)
     store.receive(.recordPermissionResponse(true)) {
       $0.audioRecorderPermission = .allowed
       $0.currentRecording = .init(
@@ -133,6 +134,7 @@ class VoiceMemosTests: XCTestCase {
       )
     }
     audioRecorderSubject.send(completion: .failure(.couldntActivateAudioSession))
+    self.mainRunLoop.advance(by: 0.5)
     store.receive(.audioRecorder(.failure(.couldntActivateAudioSession))) {
       $0.alert = .init(title: .init("Voice memo recording failed."))
       $0.currentRecording = nil
@@ -189,7 +191,7 @@ class VoiceMemosTests: XCTestCase {
   func testPlayMemoFailure() {
     var environment = VoiceMemosEnvironment.failing
     environment.audioPlayer.play = { _ in Effect(error: .decodeErrorDidOccur) }
-    environment.mainRunLoop = .immediate
+    environment.mainRunLoop = self.mainRunLoop.eraseToAnyScheduler()
 
     let url = URL(string: "https://www.pointfree.co/functions")!
     let store = TestStore(
@@ -278,7 +280,7 @@ class VoiceMemosTests: XCTestCase {
     var environment = VoiceMemosEnvironment.failing
     environment.audioPlayer.play = { _ in .none }
     environment.audioPlayer.stop = { .none }
-    environment.mainRunLoop = .immediate
+    environment.mainRunLoop = self.mainRunLoop.eraseToAnyScheduler()
 
     let store = TestStore(
       initialState: VoiceMemosState(

Sources/ComposableArchitecture/Effects/Cancellation.swift

@@ -29,18 +29,16 @@ extension Effect {
   ///     canceled before starting this new one.
   /// - Returns: A new effect that is capable of being canceled by an identifier.
   public func cancellable(id: AnyHashable, cancelInFlight: Bool = false) -> Effect {
-    let effect = Deferred { () -> Publishers.HandleEvents<PassthroughSubject<Output, Failure>> in
+    let effect = Deferred { () -> Publishers.HandleEvents<Publishers.PrefixUntilOutput<Self, PassthroughSubject<Void, Never>>> in
       cancellablesLock.lock()
       defer { cancellablesLock.unlock() }
 
-      let subject = PassthroughSubject<Output, Failure>()
-      let cancellable = self.subscribe(subject)
+      let cancellationSubject = PassthroughSubject<Void, Never>()
 
       var cancellationCancellable: AnyCancellable!
       cancellationCancellable = AnyCancellable {
         cancellablesLock.sync {
-          subject.send(completion: .finished)
-          cancellable.cancel()
+          cancellationSubject.send(())
           cancellationCancellables[id]?.remove(cancellationCancellable)
           if cancellationCancellables[id]?.isEmpty == .some(true) {
             cancellationCancellables[id] = nil
@@ -52,10 +50,11 @@ extension Effect {
         cancellationCancellable
       )
 
-      return subject.handleEvents(
-        receiveCompletion: { _ in cancellationCancellable.cancel() },
-        receiveCancel: cancellationCancellable.cancel
-      )
+      return self.prefix(untilOutputFrom: cancellationSubject)
+        .handleEvents(
+          receiveCompletion: { _ in cancellationCancellable.cancel() },
+          receiveCancel: cancellationCancellable.cancel
+        )
     }
     .eraseToEffect()
 

Tests/ComposableArchitectureTests/EffectCancellationTests.swift

@@ -286,4 +286,20 @@ final class EffectCancellationTests: XCTestCase {
     scheduler.advance(by: 1)
     XCTAssertNoDifference(expectedOutput, [])
   }
+  
+  func testNestedMergeCancellation() {
+    let effect = Effect<Int, Never>.merge(
+      (1...2).publisher
+        .eraseToEffect()
+        .cancellable(id: 1)
+    )
+    .cancellable(id: 2)
+
+    var output: [Int] = []
+    effect
+      .sink { output.append($0) }
+      .store(in: &cancellables)
+
+    XCTAssertEqual(output, [1, 2])
+  }
 }