Reactive Programming in iOS with Combine

Certain operators in Combine take in a Scheduler as an parameter. Most of the time, this is DispatchQueue.main, but it can vary. What is a scheduler in this case? Per Apple’s documentation, a scheduler is a protocol that defines when and how to execute a closure.

This is most, but not all, of the story. The word scheduler implies the time that the closure is executed, either now or in the future. What’s missing is where the closure is executed. There is no reference to which thread the closure will be executed on; this is left to the concrete implementation of the scheduler you choose. So the key message to take home here: a scheduler is not equal to a thread.

Here’s what a scheduler looks like from an event flow standpoint:

A button press occurs on the main thread, some process fires off on a background scheduler, and the results are sent back to the main thread where the app’s UI is updated. In this episode you’ll learn about 2 operators related to subscribers - subscribe and receive.

The subscribe(on:) and subscribe(on:options) operator create the subscription, or start the work, on the specified scheduler. Back in the first part of the course, you learned that a publisher is inanimate until something subscribes to it. What happens when that take place?

First, Publisher receives the subscriber and creates a subscription. Then, Subscriber receives the subscription and requests values from the publisher (these are the dotted lines in the diagram). Then the Publisher starts work via the Subscription. Then The Publisher emits values via that Subscription.

Once that happens, operators transform values (where applicable). Finally, the Subscriber receives the final value.

Steps 1, 2, and 3 typically happen on the thread that is current when the Publishers gets the subscription. However, when you use the subscribe(on:) operator, all these operations run on the scheduler you specified. An example of using subscribe(on:) is when you want the publisher to perform an expensive computation, and you want to avoid blocking the main thread. Let’s look at this in an example.

Open the Starter.playground for this episode and selected the subscribeOn-receiveOn page. Make sure the Debug area is displayed. Add an expensive publisher, which is provided by the Publishers.ExpensiveComputation(duration:) method in the playground supplmental source code. This simulates a long running computation that emits a string after the specified duration:

// 1
let computationPublisher = Publishers.ExpensiveComputation(duration: 3)

Then create a serial queue that you’ll use to trigger the computation on a specific scheduler. DispatchQueue adopts the Scheduler protocol, so this works great for our demo.

// 2
let queue = DispatchQueue(label: "serial queue")

Next, obtain the currentThread number, thanks again to an extension on Thread provided in the playground supplemental source code.

// 3
let currentThread = Thread.current.number
print("Start computation publisher on thread \(currentThread)")

Now make a subscription to the computationPublisher. Use a sink to display the value the computationPublisher emits.

let subscription = computationPublisher
  .sink { value in
    let thread = Thread.current.number
    print("Received computation result on thread \(thread): '\(value)'")
  }

Run the playground and examine the output in the debug area. The code is running on thread 1, which is the main thread. It subscribes to the publisher here. Publisher receives a subscriber. It creates a subscription, then starts work. When the work is complete, the publisher delivers the result through the subscription and completes.

Insert a subscribe(on:) operator and use the queue we created for the parameter.

let subscription = computationPublisher
  .subscribe(on: queue)
  .sink { value in...

Run he playground again. The code still starts on the main thread. This time, however, the publisher/subscriber process happens on thread 5, which is one of the threads from the queue you made. Note that you may see a different number when you run the playground; just note it is different from 1, which is the main thread.

The receive(on:) and receive(on:options:), on the other hand, deliver values on the specified scheduler. What does this mean?

What if you wanted to update your UI with updated values after this subscription completes? Typically you would have to dispatch that code back to the main thread with something like DispatchQueue.main.async, but the receive operator can take care of that for you. Let’s go back to the playground for an example.

Continue with the previous demo, and insert a receive operator after the subscribe operator, using the main queue as the parameter here.

let subscription = computationPublisher
  .subscribe(on: queue)
  .receive(on: DispatchQueue.main)
  .sink { value in

Run the playground. The difference here is that the computation results is received back on thread 1 - the main thread - instead of the background thread. With this value in hand on the main queue, you can update your user interface components safely.

In this episode, you learned about 2 scheduling operators in Combine - subscribe(on:) and receive(on:).

subscribe(on:) causes the subscription to the publisher to occur on whatever scheduler you specify. receive(on:) allows you to deliver values from the background thread to the main thread, so you can, for example, update your user interface safely.

In the next episode, you’ll learn about sequence related operators in Combine - I’ll see you then.