What do receive(on:) and subscribe(on:) do, and how do you know which thread a stage runs on?
Predict first
After receive(on:) a private queue, does the sink run on that queue?
import Combine
import Dispatch
var bag = Set<AnyCancellable>()
let q = DispatchQueue(label: "bg")
let key = DispatchSpecificKey<Bool>()
q.setSpecific(key: key, value: true)
let sem = DispatchSemaphore(value: 0)
nonisolated(unsafe) var onQ = false
Just(1).receive(on: q).sink { _ in
onQ = DispatchQueue.getSpecific(key: key) == true
sem.signal()
}.store(in: &bag)
sem.wait()
print(onQ)- true
- false
Predict first
Does receive(on: DispatchQueue.main) deliver before or after the next line of code?
import Combine
import Foundation
var bag = Set<AnyCancellable>()
var order: [String] = []
Just("x").receive(on: DispatchQueue.main).sink { _ in order.append("sink") }.store(in: &bag)
order.append("after subscribe")
try? await Task.sleep(nanoseconds: 100_000_000)
print(order)- ["sink", "after subscribe"]
- ["after subscribe", "sink"]
Predict first
With the immediate scheduler:
import Combine
var bag = Set<AnyCancellable>()
var imm: [Int] = []
Just(1).receive(on: ImmediateScheduler.shared).sink { imm.append($0) }.store(in: &bag)
print(imm)- []
- [1]
Mental model. Each stage is a worker who passes a parcel on. Nothing changes the worker unless a hop (receive(on:)) hands the parcel to a different team.
A scheduler (DispatchQueue, OperationQueue, RunLoop, ImmediateScheduler) decides when and where a closure runs. Combine does nothing to threads by default: a value is processed on the thread of whoever sends it, all the way down the chain, so a URLSession publisher’s values arrive on a background queue and a subject’s on the caller’s thread. Two operators move work. receive(on:) hops downstream work to a scheduler: everything after it (operators and the final sink) runs there. It is the one you need for UI (.receive(on: DispatchQueue.main) just before sink or assign), and it can be used several times in one chain. subscribe(on:) moves the subscription (the work of starting the upstream and requesting values) to a scheduler, which matters only for publishers that do their work on subscription, such as a Deferred or a Future doing synchronous work; it does not move the delivery of values.
Details that matter. receive(on: DispatchQueue.main) is asynchronous even when already on the main thread: the value is enqueued, so code after the subscribe call runs before the sink (order: “after subscribe”, then “sink”), and tests must wait. Use ImmediateScheduler.shared to run synchronously (useful in tests, and a no-op hop in production). A hop costs a dispatch, so do it once near the end rather than between every operator, and do heavy transforms (decoding) on a background queue before hopping to the main one. Remember RunLoop.main and Timer publishers fire in the default run-loop mode, so they stall during scrolling (see the run-loop topic). To find the thread, use the debugger’s stack or log Thread.current in handleEvents; and know that SwiftUI’s @Published updates must still happen on the main thread.
Everything downstream of receive(on:) runs on the scheduler it names.
The hop is asynchronous, so the value is delivered on a later turn even though a plain Just would be synchronous.
The immediate scheduler runs the closure synchronously, which is why it is the scheduler of choice in tests.
Common traps
- Updating UI in a
sinkwithoutreceive(on: DispatchQueue.main). The value arrives on a background queue. - Expecting
receive(on: .main)to be synchronous on the main thread. It always defers by one turn.
They might follow up with
- When does
subscribe(on:)matter? When subscribing itself does heavy work; it does not move where values are delivered. - How do you make a pipeline synchronous in a test? Use
ImmediateScheduler.sharedfor hops.