Does Swift.Combine Publishers.CombineLatest have a bug handling back pressure correctly? (demand from downstream)

Viewed 72

I am seeing a very odd behavior in Publishers.CombineLatest related to demand management. In my code I have created a custom .withLatestFrom operator which manages backpressure by limiting upstream demand.

I am running in a situation where Publishers.CombineLatest propagates demand only to the 2nd publisher (latest2) but not to the first one (latest1).

What do you think? Is this a bug in CombineLatest - or do any of you see another reason why this might happen? I am not sure if miss anything, but to me it looks like quite a serious bug in a very common operator?

I am a bit puzzled - as this situation is currently stopping a publisher chain in my app...

Here is a test case (I have simplified it from my previous post and taken out any non-standard code):

func testCombineLatest() {
    
    let latest1 = PassthroughSubject<Int,Never>()
    let latest2 = PassthroughSubject<Int,Never>()
    
    var result:[[Int]] = []
    var subscription:Subscription?
    
    let subscriber = AnySubscriber<(Int,Int),Never>(
        receiveSubscription: {sub in
            subscription = sub
            sub.request(.max(1))
        },
        receiveValue: { (v1,v2) in
            result.append([v1,v2])
            return .max(1)
        },
        receiveCompletion: {_ in}
    )
    
    let publisher = Publishers.CombineLatest(latest1.print("Latest1"), latest2.print("Latest2"))
        .print("CombineLatest")
    
    publisher
        .subscribe(subscriber)
    
    latest1.send(1)
    latest2.send(1)
    latest1.send(2) //<- This has no effect.
    latest2.send(2)
    
    latest1.send(completion: .finished)
    latest2.send(completion: .finished)
    
    print("Result is:\(result)")
    
    XCTAssertEqual(result, [[1,1], [2,1], [2,2] ])

}

And here ist the output from the test run:

Test Case '-[CAPTests.CAPTestPublisherExtensions testCombineLatest]' started.
Latest1: receive subscription: (PassthroughSubject)
Latest2: receive subscription: (PassthroughSubject)
CombineLatest: receive subscription: (CombineLatest)
CombineLatest: request max: (1)
Latest1: request max: (1)
Latest2: request max: (1)
Latest1: receive value: (1)
Latest2: receive value: (1)
CombineLatest: receive value: ((1, 1))
CombineLatest: request max: (1) (synchronous)
Latest2: request max: (1) (synchronous)
Latest2: receive value: (2)
CombineLatest: receive value: ((1, 2))
CombineLatest: request max: (1) (synchronous)
Latest2: request max: (1) (synchronous)
Latest1: receive finished
Latest2: receive finished
CombineLatest: receive finished
Result is:[[1, 1], [1, 2]]
error: -[CAPTests.CAPTestPublisherExtensions testCombineLatest] : XCTAssertEqual failed: ("[[1, 1], [1, 2]]") is not equal to ("[[1, 1], [2, 1], [2, 2]]")
0 Answers
Related