title: Swift Combine - Publishers.CombineLatest on multiple Threads
body: <p>When using Publishers.CombineLatest with Publishers which run an threads other that Main the .sink of the <code>Publishers.CombineLatest</code> is not always called.</p>
  
  <p>The issue does not appear every time this is why I created unit tests which try the test 100 times in a row. Usually they fail after 4-5 iterations.</p>
  
  <pre><code>import XCTest
  import Combine
  
  class CombineLatestTests: XCTestCase {
  
      override func setUp() {
          continueAfterFailure = false
      }
  
      func testCombineLatest_receiveOn() {
          for x in 0...1000 {
              print("---------- RUN \(x)")
              let queue1 = DispatchQueue.global(qos: .userInitiated)
              let queue2 = DispatchQueue.global(qos: .background)
  
              let subj1 = PassthroughSubject&lt;Int, Never&gt;()
              let subj2 = PassthroughSubject&lt;Int, Never&gt;()
  
              let publ1 = subj1.receive(on: queue1).map { value -&gt; Int in
                  print("-- Observer 1: \(value), Thread: \(Thread.current)")
                  return value
              }
              let publ2 = subj2.receive(on: queue2).map { value -&gt; Int in
                  print("-- Observer 2: \(value), Thread: \(Thread.current)")
                  return value
              }
  
              let exp = expectation(description: "expect values")
              exp.assertForOverFulfill = false
              let canc = Publishers.CombineLatest(publ1, publ2)
                  .sink { value1, value2 in
                      print("-- recieved \(value1):\(value2) on \(Thread.current)")
                      if value1 == 10, value2 == 20 {
                          exp.fulfill()
                      }
                  }
  
              subj1.send(5)
              subj2.send(20)
              subj1.send(10)
  
              wait(for: [exp], timeout: 10)
              canc.cancel()
          }
      }
  
      func testCombineLatest_currentValue_receiveOn() {
          for x in 0...100 {
              print("---------- RUN \(x)")
              let queue1 = DispatchQueue.global(qos: .userInitiated)
              let queue2 = DispatchQueue.global(qos: .background)
  
              let subj1 = CurrentValueSubject&lt;Int, Never&gt;(0)
              let subj2 = CurrentValueSubject&lt;Int, Never&gt;(0)
  
              let publ1 = subj1.receive(on: queue1).map { value -&gt; Int in
                  print("-- Observer 1: \(value), Thread: \(Thread.current)")
                  return value
              }
              let publ2 = subj2.receive(on: queue2).map { value -&gt; Int in
                  print("-- Observer 2: \(value), Thread: \(Thread.current)")
                  return value
              }
  
              let exp = expectation(description: "expect values")
              exp.assertForOverFulfill = false
              let canc = Publishers.CombineLatest(publ1,
                                                  publ2)
                  .sink { value1, value2 in
                      print("-- recieved \(value1):\(value2) on \(Thread.current)")
                      if value1 == 10, value2 == 20 {
                          exp.fulfill()
                      }
                  }
  
              subj1.send(10)
              subj2.send(20)
  
              wait(for: [exp], timeout: 3)
              canc.cancel()
          }
      }
  
      func testCombineLatest_subscribeOn() {
          for x in 0...100 {
              print("---------- RUN \(x)")
              let queue1 = DispatchQueue.global(qos: .userInitiated)
              let queue2 = DispatchQueue.global(qos: .background)
  
              let subj1 = PassthroughSubject&lt;Int, Never&gt;()
              let subj2 = PassthroughSubject&lt;Int, Never&gt;()
  
              let publ1 = subj1.map { value -&gt; Int in
                  print("-- Observer 1: \(value), Thread: \(Thread.current)")
                  return value
              }
              let publ2 = subj2.map { value -&gt; Int in
                  print("-- Observer 2: \(value), Thread: \(Thread.current)")
                  return value
              }
  
              let exp = expectation(description: "expect values")
              exp.assertForOverFulfill = false
              let canc = Publishers.CombineLatest(publ1, publ2)
                  .sink { value1, value2 in
                      print("-- recieved \(value1):\(value2) on \(Thread.current)")
                      if value1 == 10, value2 == 20 {
                          exp.fulfill()
                      }
                  }
  
              queue1.async {
                  subj1.send(5)
                  subj1.send(10)
              }
  
              queue2.async {
                  subj2.send(20)
              }
  
              wait(for: [exp], timeout: 5)
              canc.cancel()
          }
      }
  
  }
  
  </code></pre>
  
  <p>Here are the logs of the 3rd test</p>
  
  <pre><code>Test Case '-[xxxx.CombineLatestTests testCombineLatest_currentValue_receiveOn]' started.
  ---------- RUN 0
  -- Observer 2: 0, Thread: &lt;NSThread: 0x6000004e0f80&gt;{number = 9, name = (null)}
  -- Observer 1: 0, Thread: &lt;NSThread: 0x6000004f0000&gt;{number = 7, name = (null)}
  -- Observer 2: 20, Thread: &lt;NSThread: 0x6000004f6e00&gt;{number = 6, name = (null)}
  -- recieved 0:0 on &lt;NSThread: 0x6000004f0000&gt;{number = 7, name = (null)}
  -- Observer 1: 10, Thread: &lt;NSThread: 0x600000439880&gt;{number = 4, name = (null)}
  -- recieved 10:20 on &lt;NSThread: 0x600000439880&gt;{number = 4, name = (null)}
  ---------- RUN 1
  -- Observer 2: 0, Thread: &lt;NSThread: 0x6000004f0000&gt;{number = 7, name = (null)}
  -- Observer 1: 0, Thread: &lt;NSThread: 0x6000004f6e00&gt;{number = 6, name = (null)}
  -- Observer 2: 20, Thread: &lt;NSThread: 0x6000004e0f80&gt;{number = 9, name = (null)}
  -- recieved 0:0 on &lt;NSThread: 0x6000004f6e00&gt;{number = 6, name = (null)}
  -- Observer 1: 10, Thread: &lt;NSThread: 0x6000004e80c0&gt;{number = 10, name = (null)}
  -- recieved 10:20 on &lt;NSThread: 0x6000004e80c0&gt;{number = 10, name = (null)}
  ---------- RUN 2
  -- Observer 2: 0, Thread: &lt;NSThread: 0x6000004f6e00&gt;{number = 6, name = (null)}
  -- Observer 1: 0, Thread: &lt;NSThread: 0x6000004e0f80&gt;{number = 9, name = (null)}
  -- Observer 2: 20, Thread: &lt;NSThread: 0x6000004e80c0&gt;{number = 10, name = (null)}
  -- recieved 0:0 on &lt;NSThread: 0x6000004e0f80&gt;{number = 9, name = (null)}
  -- Observer 1: 10, Thread: &lt;NSThread: 0x600000439880&gt;{number = 4, name = (null)}
  -- recieved 10:20 on &lt;NSThread: 0x600000439880&gt;{number = 4, name = (null)}
  ---------- RUN 3
  -- Observer 2: 0, Thread: &lt;NSThread: 0x600000439880&gt;{number = 4, name = (null)}
  -- Observer 1: 0, Thread: &lt;NSThread: 0x6000004e0f80&gt;{number = 9, name = (null)}
  -- Observer 2: 20, Thread: &lt;NSThread: 0x6000004f6e00&gt;{number = 6, name = (null)}
  -- recieved 0:0 on &lt;NSThread: 0x6000004e0f80&gt;{number = 9, name = (null)}
  -- Observer 1: 10, Thread: &lt;NSThread: 0x6000004e80c0&gt;{number = 10, name = (null)}
  -- recieved 10:20 on &lt;NSThread: 0x6000004e80c0&gt;{number = 10, name = (null)}
  ---------- RUN 4
  -- Observer 1: 0, Thread: &lt;NSThread: 0x6000004f6e00&gt;{number = 6, name = (null)}
  -- Observer 2: 0, Thread: &lt;NSThread: 0x6000004f0000&gt;{number = 7, name = (null)}
  -- recieved 0:0 on &lt;NSThread: 0x6000004f0000&gt;{number = 7, name = (null)}
  -- Observer 1: 10, Thread: &lt;NSThread: 0x600000439880&gt;{number = 4, name = (null)}
  -- Observer 2: 20, Thread: &lt;NSThread: 0x6000004e80c0&gt;{number = 10, name = (null)}
  -- recieved 10:0 on &lt;NSThread: 0x600000439880&gt;{number = 4, name = (null)}
  CombineLatestTests.swift:93: error: : Asynchronous wait failed: Exceeded timeout of 3 seconds, with unfulfilled expectations: "expect values".
  Test Suite 'CombineLatestTests' failed at 2020-03-04 20:37:24.957.
       Executed 3 tests, with 3 failures (0 unexpected) in 18.159 (18.161) seconds
  </code></pre>