Repository: broadwaylamb/OpenCombine (swift)

Issue:
Crash when deallocating a manually connected multicast before the Cancellable
## What did you do?

Deallocated a manually connected multicast subject, before the `Cancellable` is released.

## What did you expect to happen?

The deallocation to be handled cleanly.

## What happened instead?

The running Application crashes with an `EXC_BAD_INSTRUCTION` due to an `UnfairLock` being recursively entered within `SubjectSubscriber.swift`.

## Environment

**OpenCombine version:** 0.10.1
**Xcode version:** Any
**Swift version:** Any
**Platform(s) running OpenCombine:** iOS

## Example code

```swift
        var originalSubject: PassthroughSubject<Void, Never>? = PassthroughSubject<Void, Never>()
        var newSubject: PassthroughSubject<Void, Never>? = PassthroughSubject<Void, Never>()
        var cancellable: Cancellable? = originalSubject?.multicast(subject: newSubject!).connect()

        newSubject = nil
        originalSubject = nil
        cancellable = nil
```

## Notes

You can work around this by ensuring your `Cancellable` is released before the subjects.

Pull request description:
This is how it is in Combine.

Fixes #186.

Patch:
diff --git a/Sources/OpenCombine/Helpers/SubjectSubscriber.swift b/Sources/OpenCombine/Helpers/SubjectSubscriber.swift
index ad6306d3c..1010850ce 100644
--- a/Sources/OpenCombine/Helpers/SubjectSubscriber.swift
+++ b/Sources/OpenCombine/Helpers/SubjectSubscriber.swift
@@ -5,7 +5,6 @@
 //  Created by Sergej Jaskiewicz on 16/09/2019.
 //
 
-// NOTE: This class has been audited for thread safety.
 internal final class SubjectSubscriber<Downstream: Subject>
     : Subscriber,
       CustomStringConvertible,
@@ -14,7 +13,7 @@ internal final class SubjectSubscriber<Downstream: Subject>
       Subscription
 {
     private let lock = UnfairLock.allocate()
-    private var downstreamSubject: Downstream?
+    private weak var downstreamSubject: Downstream?
     private var upstreamSubscription: Subscription?
 
     private var isCancelled: Bool { return downstreamSubject == nil }


Test patch:
diff --git a/Tests/OpenCombineTests/PublisherTests/MulticastTests.swift b/Tests/OpenCombineTests/PublisherTests/MulticastTests.swift
index efc1ce189..210c5cb15 100644
--- a/Tests/OpenCombineTests/PublisherTests/MulticastTests.swift
+++ b/Tests/OpenCombineTests/PublisherTests/MulticastTests.swift
@@ -146,6 +146,8 @@ final class MulticastTests: XCTestCase {
                                          .value(7),
                                          .completion(.finished),
                                          .subscriber])
+
+        connection.cancel()
     }
 
     func testSubscribeAfterCompletion() {
@@ -288,6 +290,19 @@ final class MulticastTests: XCTestCase {
         }
     }
 
+    func testDeallocateConnectedMulticastBeforeCancelling() {
+        // https://github.com/OpenCombine/OpenCombine/issues/186
+
+        let publisher = CustomPublisher(subscription: CustomSubscription())
+        var subjectDestroyed = false
+        var subject: TrackingSubject<Int>? = .init(onDeinit: { subjectDestroyed = true })
+
+        let connection = publisher.multicast(subject: subject!).connect()
+        subject = nil
+        XCTAssertTrue(subjectDestroyed)
+        connection.cancel() // Shouldn't recursively aquire a non-recursive lock here.
+    }
+
     func testLazySubjectCreation() {
         let publisher = PassthroughSubject<Int, TestingError>()
         var counter = 0
@@ -406,10 +421,12 @@ final class MulticastTests: XCTestCase {
         publisher.send(42)
         publisher.send(completion: .finished)
 
-        XCTAssertEqual(tracking.history, [.subscription("Multicast"),
-                                          .value(42),
-                                          .value(42),
-                                          .completion(.finished)])
+        withExtendedLifetime(multicast) {
+            XCTAssertEqual(tracking.history, [.subscription("Multicast"),
+                                              .value(42),
+                                              .value(42),
+                                              .completion(.finished)])
+        }
 
         connection1.cancel()
         connection2.cancel()