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]]")