Pipe

final class Pipe<Element> {
    private(set) var send: (Element) -> Void = { _ in }
    func makeStream() -> AsyncStream<Element> {
        .init(Element.self, bufferingPolicy: .bufferingNewest(1)) { continuation in
            send = { element in
                continuation.yield(element)
            }
        }
    }
}

final class ThrowingPipe<Element> {
    private(set) var send: (Result<Element, Error>) -> Void = { _ in }
    var failure: Error?
    func makeStream() -> AsyncThrowingStream<Element, Error> {
        .init(Element.self, bufferingPolicy: .bufferingNewest(1)) { [weak self] continuation in
            send = { result in
                result.mapError { someError -> Error in
                    let failure = self?.failure ?? someError
                    self?.failure = failure
                    return failure
                }
                guard  let failure = self?.failure else {
                    continuation.yield(with: result)
                    return
                }
                continuation.yield(with: .failure(failure))
            }
        }
    }
    func send(_ element: Element) {
        send(.success(element))
    }
}