AsyncPassthroughSubject - Swift Structured Concurrency

AsyncPassthroughSubject.swift

import Foundation

/// Provides a safe stream of `values`.
public struct AsyncPassthroughSubject<T> {
    public let values: AsyncStream<T>
    private let input: AsyncStream<T>.Continuation
    private var isFinished = false
    
    public init(
        bufferingPolicy: AsyncStream<T>.Continuation.BufferingPolicy = .unbounded
    ) {
        var input: AsyncStream<T>.Continuation!
        self.values = AsyncStream<T>(bufferingPolicy: bufferingPolicy) { input = $0 }
        self.input = input
    }
    
    public func send(_ value: T) {
        guard !isFinished else { return }
        input.yield(value)
    }
    
    public mutating func finish() {
        isFinished = true
        input.finish()
    }
}