Created
November 17, 2025 15:44
-
-
Save jdanthinne/f45eb390102b03e0244a6519e9b4382c to your computer and use it in GitHub Desktop.
Extension for AsyncSequence to launch an operation when stream starts
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import Foundation | |
| package extension AsyncSequence where Element: Sendable, Self: Sendable { | |
| func onSubscribed(perform: @Sendable @escaping () async throws -> Void) -> AsyncThrowingStream<Element, Error> { | |
| .init { continuation in | |
| let task = Task { | |
| do { | |
| // Use a task group to run both operations concurrently | |
| try await withThrowingTaskGroup(of: Void.self) { group in | |
| // Launch the iteration task | |
| group.addTask { | |
| for try await element in self { | |
| guard !Task.isCancelled else { return } | |
| continuation.yield(element) | |
| } | |
| } | |
| // Launch the perform block | |
| group.addTask { | |
| try await perform() | |
| } | |
| // Wait for both tasks to complete | |
| try await group.waitForAll() | |
| } | |
| continuation.finish() | |
| } catch { | |
| continuation.finish(throwing: error) | |
| } | |
| } | |
| continuation.onTermination = { _ in task.cancel() } | |
| } | |
| } | |
| } |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment