🆕 Add a 'collect(until:)' operator

Author: ferologics

Sources/Operators/CollectUntilTrigger.swift

@@ -0,0 +1,51 @@
+//
+//  CollectUntilTrigger.swift
+//  CombineExt
+//
+//  Created by ferologics on 09/06/2021.
+//  Copyright © 2021 Combine Community. All rights reserved.
+//
+
+#if canImport(Combine)
+import Combine
+
+@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+public extension Publisher {
+    func collect<CollectionTrigger>(
+        until trigger: CollectionTrigger
+    ) -> AnyPublisher<[Output], Failure> where
+        CollectionTrigger: Publisher,
+        CollectionTrigger.Output == Void,
+        CollectionTrigger.Failure == Never {
+        var events = [Output]()
+
+        let eventPublisher = PassthroughSubject<[Output], Failure>()
+        var cancellables = [AnyCancellable]()
+
+        self.sink { completion in
+            eventPublisher.send(completion: completion)
+        } receiveValue: { output in
+            events.append(output)
+        }
+        .store(in: &cancellables)
+
+        trigger.sink { _ in
+            eventPublisher.send(events)
+            events = []
+        }
+        .store(in: &cancellables)
+
+        func cleanUp() {
+            cancellables.forEach { $0.cancel() }
+            cancellables = []
+        }
+
+        return eventPublisher
+            .handleEvents(
+                receiveCompletion: { _ in cleanUp() },
+                receiveCancel: { cleanUp() }
+            )
+            .eraseToAnyPublisher()
+    }
+}
+#endif

Tests/CollectUntilTriggerTests.swift

@@ -0,0 +1,41 @@
+//
+//  CollectUntilTriggerTests.swift
+//  CombineExtTests
+//
+//  Created by ferologics on 09/06/2021.
+//
+
+#if !os(watchOS)
+import XCTest
+import Combine
+
+@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+class CollectUntilTriggerTests: XCTestCase {
+    var subscription: AnyCancellable!
+
+    func test() {
+        // Given
+        let elements = [1,2,3,4,5]
+        var receivedElements = [Int]()
+        let elementsPublisher = PassthroughSubject<Int, Never>()
+        let trigger = PassthroughSubject<Void, Never>()
+
+        // When
+        subscription = elementsPublisher
+            .collect(until: trigger)
+            .sink { receivedElements = $0 }
+
+        for x in elements {
+            elementsPublisher.send(x)
+        }
+
+        // Then
+        XCTAssertTrue(receivedElements.isEmpty)
+        trigger.send(())
+        XCTAssertEqual(elements.count, receivedElements.count)
+        for (a, b) in zip(elements, receivedElements) {
+            XCTAssertEqual(a, b)
+        }
+    }
+}
+#endif