Combine `.receive(on: RunLoop.main)` loses sent value. How can I make it work?

Using Swift
20

There's a lot here, and I'll try to answer as best I can.

receive(on:) will cause execution of all receive messages (receive(subscription:), receive(_ input: Input), receive(completion:)) on the specified Scheduler (a run loop in this case).

Ordering of messages in Combine follows a strict rule:

  • 1 subscription downstream
  • Request of N values upstream
  • N values downstream
  • 1 completion (or no completion, if infinite)

Nothing can be sent before the subscription is received downstream, because the downstream did not have an opportunity to make a request yet. Requesting values requires a subscription. The subscription arrives asynchronously (because that is what receive(on:) is defined to do, and also because everything in Combine is designed to be potentially asynchronous). In the example, the RunLoop has not yet run and therefore there is no possibility of a subscription being received by the sink operator.

Here is another way to "fix" it:

let control = PassthroughSubject<Int,Never>()
let pipe = control.receive(on: RunLoop.main).sink(receiveValue: { print($0) })
RunLoop.main.run(until: Date().addingTimeInterval(0.1)) // run the run loop for .1 seconds
control.send(233)
RunLoop.main.run() // run the run loop forever

The reason the first run loop run is required is because this code is already executing on the main thread and therefore blocks. RunLoop needs a chance to execute the enqueued work of delivering the subscription.

I think also it may help to remember that the chain of operators you're setting up are descriptions; not much is happening when you call receive(on:) besides recording your arguments in a struct. The actual "instantiation" of the chain happens when the last thing in the chain calls subscribe. sink is different than the others:

    public func sink(
        receiveCompletion: @escaping ((Subscribers.Completion<Failure>) -> Void),
        receiveValue: @escaping ((Output) -> Void))
        -> AnyCancellable 
    {
        let sink = Subscribers.Sink<Self.Output, Self.Failure>(receiveCompletion: receiveCompletion, receiveValue: receiveValue)
        self.subscribe(sink)
        return AnyCancellable(sink)
    }

So once you call it, it calls subscribe on receive(on:). In response to that, receive(on:) says "ok, here is your subscription" and sends is asynchronously like this (simplified):

scheduler.schedule(options: options) {
    ds.receive(subscription: self)
}

And when sink receives that subscription it asks it for some values:

public func receive(subscription: Subscription) {
    subscription.request(.unlimited)
}

Once the unlimited request comes in, receive sends it to the subject, at which point values can be sent safely.

Hopefully this helps clarify.

3 Likes