[Bug]The Publishers.ReceiveOn may lead to subscriber never receive the published signal when scheduler is concurrent queue.

[dito010]

For Future type publisher, it send signal then send completion to the Publishers.ReceiveOn.

In Publishers.ReceiveOn:

for signal:

scheduler.schedule(options: options) {
   self.scheduledReceive(input)
}

for completion:

scheduler.schedule(options: options) {
   self.scheduledReceive(completion: completion)
}

but the scheduler may execute the completion command before signal, in this case, the state.subscription is nil then returned.

private func scheduledReceive(_ input: Input) {
  lock.lock()
  guard state.subscription != nil else {
    lock.unlock()
    return
  }
   
   // ...
}

This result is not match expectations.

And the Apple's Combine has the same bug yet 😂.

We should promise the subscriber will always receive the signal?

Some solution code:

// for signal.
lockedActions.write().safeRun({
    $0.pointee.insert({ [weak self] in self?.scheduledReceive(input) }, at: 0)
})
scheduler.schedule(options: options) {
    self.lockedActions.write().safeRun({ $0.pointee.popLast()?() })
}

// for complection.
lockedActions.write().safeRun({
    $0.pointee.insert({ [weak self] in self?.scheduledReceive(completion: completion) }, at: 0)
})
scheduler.schedule(options: options) {
    self.lockedActions.write().safeRun({ $0.pointee.popLast()?() })
}