Swift Combine framework publisher that allows you to finish/cancel/close the stream based on its contents.

finishable.swift

public struct FinishablePublisher<Upstream>: Publisher where Upstream: Publisher {
    public typealias Output = Upstream.Output
    public typealias Failure = Upstream.Failure

    private let upstream: Upstream
    private let inspector: (Output) -> Bool

    init(upstream: Upstream, shouldFinish inspector: @escaping (Output) -> Bool) {
        self.upstream = upstream
        self.inspector = inspector
    }

    public func receive<S>(subscriber: S) where S: Subscriber, Failure == S.Failure, Output == S.Input {
        let shared = upstream.share()

        let finished = shared
            .filter(inspector)
            .prefix(1)

        let output = shared
            .prefix { !inspector($0) }
            .merge(with: finished)

        output.subscribe(subscriber)
    }
}

public extension Publisher {
    /// Finishes the stream _after_ the condition is met.
    /// The output that determined that will be emitted on the stream, if you don't want that value included use `.prefix` instead.
    func finish(after condition: @escaping (Output) -> Bool) -> FinishablePublisher<Self> {
        FinishablePublisher(upstream: self, shouldFinish: condition)
    }
}