问在Swift组合中不丢弃值和不平衡发布的优雅combineLatest方法

EN

extension Publisher {
    func combineLatestOptional<Other: Publisher>(_ other: Other) 
        -> AnyPublisher<(Output?, Other.Output?), Failure> 
        where Other.Failure == Failure {
        
        self.map { Optional.some($0) }.prepend(nil)
            .combineLatest(
                other.map { Optional.some($0) }.prepend(nil)
            )
            .dropFirst() // drop the first (nil, nil)
            .eraseToAnyPublisher()
    }
}

fileprivate enum State<L, R> {
    case initial         // before any one publisher emitted
    case left([L])       // left emitted; right hasn't emitted
    case right([R])      // right emitted; left hasn't emitted
    case final([L], [R]) // final steady-state
}

extension Publisher {
   func combineLatestLossless<Other: Publisher>(_ other: Other) 
          -> AnyPublisher<(Output, Other.Output), Failure> 
          where Failure == Other.Failure {

      self.combineLatestOptional(other)
          .scan(State<Output, Other.Output>.initial, { state, tuple in
             switch (state, tuple.0, tuple.1) {
             case (.initial, let l?, nil):       // left emits first value
                return .left([l])                // -> collect left values

             case (.initial, nil, let r?):       // right emits first value
                return .right([r])               // -> collect right values

             case (.left(let ls), let l?, nil):  // left emits another
                return .left(ls + [l])           // -> append to left values

             case (.right(let rs), nil, let r?): // right emits another
                return .right(rs + [r])          // -> append to right values

             case (.left(let ls), _, let r?):    // right emits after left
                return .final(ls, [r])           // -> go to steady-state

             case (.right(let rs), let l?, _):   // left emits after right
                return .final([l], rs)           // -> go to steady-state

             case (.final, let l?, let r?):      // final steady-state
                return .final([l], [r])          // -> pass the values as-is
             default:
                fatalError("shouldn't happen")
             }
         })
         .flatMap { status -> AnyPublisher<(Output, Other.Output), Failure> in
            if case .final(let ls, let rs) = status {
               return ls.flatMap { l in rs.map { r in (l, r) }}
                   .publisher
                   .setFailureType(to: Failure.self)
                   .eraseToAnyPublisher()
            } else {
               return Empty().eraseToAnyPublisher()
            }
         }
         .eraseToAnyPublisher()
   }
}

let c = pub1.combineLatestLossless(pub2)
            .sink { print($0) }

import UIKit
import Combine

func delay(_ delay:Double, closure:@escaping ()->()) {
    let when = DispatchTime.now() + delay
    DispatchQueue.main.asyncAfter(deadline: when, execute: closure)
}

class ViewController: UIViewController {
    var storage = Set<AnyCancellable>()
    let s1 = PassthroughSubject<Int,Never>()
    let s2 = PassthroughSubject<String,Never>()
    override func viewDidLoad() {
        super.viewDidLoad()
        
        let p1 = s1
        let p2 = s2.flatMap { (val:String) -> AnyPublisher<String,Never> in
            let seq = Array(repeating: val, count: 100)
            return seq.publisher.eraseToAnyPublisher()
        }
        p1.zip(p2)
            .sink{print($0)}
            .store(in: &storage)
        
        delay(1) {
            self.s1.send(1)
        }
        delay(2) {
            self.s1.send(2)
        }
        delay(3) {
            self.s1.send(3)
        }
        delay(2.5) {
            self.s2.send("X")
        }
    }
}

(1, "X")
(2, "X")
(3, "X")

var subject1 = Timer.publish(every: 1, on: .main, in: .default, options: nil)
    .autoconnect()
    .measureInterval(using: RunLoop.main, options: nil)
    .scan(DateInterval()) { res, interval in
        .init(start: res.start, duration: res.duration + interval.magnitude)
    }
    .map(\.duration)
    .map { Int($0) }
    .eraseToAnyPublisher()
var subject2 = PassthroughSubject<String, Never>()

let bufferedSubject1 = Publishers.Concatenate(prefix: Publishers.PrefixUntilOutput(upstream: subject1, other: subject2).collect().flatMap(\.publisher),
                                              suffix: subject1)

let cancel = bufferedSubject1.combineLatest(subject2)
    .sink(receiveCompletion: { c in
        print(c)
    }, receiveValue: { v in
        print(v)
    })

subject2.send("X")

DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
    subject2.send("Y")
}

(1, "X")
(2, "X")
(3, "X")
(3, "Y")
(4, "Y")
(5, "Y")
(6, "Y")

import PlaygroundSupport
import Combine

PlaygroundPage.current.needsIndefiniteExecution = true

let queue = DispatchQueue(label: "com.foo.bar")
let cancellable = letters
    .combineLatest(indices
                    .collect(.byTimeOrCount(queue, .seconds(1), .max))
                    .flatMap { indices in indices.publisher })
    .sink { letter, index in print("(\(index), \(letter))") }

indices.send(1)
DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
    indices.send(2)
    indices.send(3)
}
DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
    letters.send("X")
}

DispatchQueue.main.asyncAfter(deadline: .now() + 3.3) {
    indices.send(4)
}

DispatchQueue.main.asyncAfter(deadline: .now() + 3.5) {
    letters.send("Y")
}

DispatchQueue.main.asyncAfter(deadline: .now() + 3.7) {
    indices.send(5)
    indices.send(6)
}

(X, 1)
(X, 2)
(X, 3)
(Y, 3)
(Y, 4)
(Y, 5)
(Y, 6)