Updated ReplaySubject to replay its buffer to a new subscription while locked to ensure new values are sent after the replayed values. (#138)

Author: mruston-heb

Sources/Subjects/ReplaySubject.swift

@@ -89,21 +89,15 @@ public final class ReplaySubject<Output, Failure: Error>: Subject {
             self?.completeSubscriber(withIdentifier: subscriberIdentifier)
         }
 
-        let buffer: [Output]
-        let completion: Subscribers.Completion<Failure>?
-
         do {
             lock.lock()
             defer { lock.unlock() }
 
+            subscription.replay(buffer, completion: completion)
             subscriptions.append(subscription)
-
-            buffer = self.buffer
-            completion = self.completion
         }
 
         subscriber.receive(subscription: subscription)
-        subscription.replay(buffer, completion: completion)
     }
 
     private func completeSubscriber(withIdentifier subscriberIdentifier: CombineIdentifier) {

Tests/ReplaySubjectTests.swift

@@ -391,5 +391,38 @@ final class ReplaySubjectTests: XCTestCase {
 
         XCTAssertTrue(subject.subscriptions.isEmpty)
     }
+
+    func testReplayOrderThreadSafety() async {
+        continueAfterFailure = false
+        // Loop to ensure any race condition is caught.
+        for _ in 0..<5000 {
+            let replaySubject = ReplaySubject<Int, Never>(bufferSize: 3)
+            replaySubject.send(1)
+            replaySubject.send(2)
+
+            // Use tasks to create a new subscription on one thread
+            // while sending the third value on another thread.
+            // The new subscription should always receive [1, 2, 3]
+            await withTaskGroup(of: Void.self) { taskGroup in
+                taskGroup.addTask {
+                    let output: [Int] = await withCheckedContinuation { continuation in
+                        var cancellable: AnyCancellable?
+                        cancellable = replaySubject.collect(3).first().sink(receiveValue: { value in
+                            continuation.resume(returning: value)
+                            withExtendedLifetime(cancellable) { cancellable = nil }
+                        })
+                    }
+
+                    await MainActor.run {
+                        XCTAssertEqual(output, [1, 2, 3])
+                    }
+                }
+
+                taskGroup.addTask {
+                    replaySubject.send(3)
+                }
+            }
+        }
+    }
 }
 #endif