Pipeline.swift

import Combine


precedencegroup SinkPrecedence { }
precedencegroup FilterPrecedence {
    associativity: left
    higherThan: SinkPrecedence
}

infix operator | : FilterPrecedence
infix operator > : SinkPrecedence
infix operator &> : SinkPrecedence

extension Publisher {
    
    static func | <U>(lhs: Self, rhs: @escaping (Output) async throws -> U) -> some Publisher {
        lhs.mapError { $0 as Error }
            .flatMap { value in
                Future<U, Error> { promise in
                    Task {
                        do {
                            promise(.success(try await rhs(value)))
                        } catch {
                            promise(.failure(error))
                        }
                    }
                }
            }
    }
    
    static func > (lhs: Self, rhs: @escaping (Output) -> Void) -> some Cancellable {  
        var sub: AnyCancellable?
        let pipeline = lhs.sink { _ in
            sub = nil
        } receiveValue: {
            rhs($0)
        }
        sub = pipeline
        return pipeline
    }
    
    static func &> (lhs: Self, rhs: @escaping (Result<Output, Error>) -> Void) -> some Cancellable {    
        var sub: AnyCancellable?
        let pipeline = lhs.sink { 
            if case .failure(let error) = $0 {
                rhs(.failure(error))
            }
            sub = nil
        } receiveValue: {
            rhs(.success($0))
        }
        sub = pipeline
        return pipeline
    }
}