Fix resource leak in ReplaySubject (#802)

Co-authored-by: Samuel Défago <[email protected]>

Sources/Core/ReplaySubject.swift

@@ -13,9 +13,9 @@ import Foundation
 /// any relevant completion.
 public final class ReplaySubject<Output, Failure>: Subject where Failure: Error {
     private let buffer: LimitedBuffer<Output>
-    private var subscriptions: [ReplaySubscription<Output, Failure>] = []
     private var completion: Subscribers.Completion<Failure>?
     private let lock = NSRecursiveLock()
+    var subscriptions: [ReplaySubscription<Output, Failure>] = []
 
     /// Creates a subject able to buffer the provided number of values.
     ///
@@ -54,6 +54,10 @@ public final class ReplaySubject<Output, Failure>: Subject where Failure: Error
     public func receive<S>(subscriber: S) where S: Subscriber, Failure == S.Failure, Output == S.Input {
         withLock(lock) {
             let subscription = ReplaySubscription(subscriber: subscriber, values: buffer.values)
+            subscription.onCancel = { [weak self] in
+                guard let self else { return }
+                subscriptions.removeAll { $0 === subscription }
+            }
             buffer.values.forEach { value in
                 subscription.append(value)
                 subscription.send()

Sources/Core/ReplaySubscription.swift

@@ -8,6 +8,8 @@ import Combine
 import Foundation
 
 final class ReplaySubscription<Output, Failure>: Subscription where Failure: Error {
+    var onCancel: (() -> Void)?
+
     private var subscriber: AnySubscriber<Output, Failure>?
     private var buffer = DemandBuffer<Output>()
     private var pendingValues: [Output] = []
@@ -22,6 +24,8 @@ final class ReplaySubscription<Output, Failure>: Subscription where Failure: Err
 
     func cancel() {
         subscriber = nil
+        onCancel?()
+        onCancel = nil
     }
 
     func append(_ value: Output) {

Tests/CoreTests/ReplaySubjectTests.swift

@@ -65,6 +65,15 @@ final class ReplaySubjectTests: XCTestCase {
         expectEqualPublished(values: [2, 3], from: subject, during: .milliseconds(100))
     }
 
+    func testSubscriptionRelease() {
+        let subject = ReplaySubject<Int, Never>(bufferSize: 1)
+        subject.send(1)
+
+        _ = subject.sink { _ in }
+
+        expect(subject.subscriptions).to(beEmpty())
+    }
+
     func testNewValuesWithMultipleSubscribers() {
         let subject = ReplaySubject<Int, Never>(bufferSize: 2)
         subject.send(1)