Getting started with Combine

2020-05-18

1. Introduction
2. Environment
3. Publisher
4. Subscriber
5. How to combine Publisher and Subscriber
6. Scheduler
7. Operators
8. Conclusions
9. References

Introduction

Combine is the Swift declarative framework for processing values over time. It enforces the reactive functional paradigm of programming, which is different from the object-oriented one prevalent in the iOS development community.

Reactive means programming with asynchronous streams of values. Functional programming is about programming with functions. In Swift, functions can be passed as arguments to other functions, returned from functions, stored in variables and data structures, and constructed at runtime as lambdas (closures). Declarative programming style means that you describe what the program does, without describing the control flow. And in an imperative style, it describes how the program works by implementing and handling a series of tasks. Imperative programs depend mainly on the state, which is usually modified with appropriations.

Programming with the Swift Combine framework is declarative, reactive and functional. It involves chaining functions and passing values ​​from one to another. This creates streams of values, which flow from input to output.

We can imagine that Combine works like this:

Environment

This tutorial is written using the following environment:

• Hardware: MacBook Pro 15' (2.5 GHz Intel Core i7, 16GB DDR3, mid-2015)
• Operating system: macOS Catalina 10.15
• Software versions:
  • Xcode: 11
  • iOS SDK: 13.4

It is important to know that to try SwiftUI you need to install Mac OS Catalina 10.15 and Xcode 11.

Publisher

Publisher sends sequences of values over time to one or more subscribers.

Publishers must comply with the following protocol:

protocol Publisher {
    associatedtype Output
    associatedtype Failure : Error
    func receive<S>(subscriber: S) where S : Subscriber, Self.Failure == S.Failure, Self.Output == S.Input
}

Publisher can submit values ​​or finish with success or error. Output defines what type of values ​​a Publisher can send. Failure defines the type of error that can fail.

The receive(subscriber: ) method connects a subscriber to a Publisher. Defines the contract: Publisher's output must match Subscriber's input and so do the error types.

Subscriber

Subscriber receives values from a Publisher.

Combine subscribers comply with the following protocol:

public protocol Subscriber : CustomCombineIdentifierConvertible {
    associatedtype Input
    associatedtype Failure : Error

    func receive(subscription: Subscription)
    func receive(_ input: Self.Input) -> Subscribers.Demand
    func receive(completion: Subscribers.Completion<Self.Failure>)
}

Subscriber can receive a value of type Input or a completion event with success or failure (Failure).

The three receive methods describe different steps of the Subscriber lifecycle.

How to combine Publisher and Subscriber

Combine has two built-in subscribers: Subscribers.Sink and Subscribers.Assign. You can connect them by calling any of these methods in a Publisher:

• sink(receiveCompletion:receiveValue:) to handle the new element or completion event in a closure (the lambda function).
• assign(to:on:) to write a new element to a property.

// 1
let publisher = Just(1)
// 2
publisher.sink(receiveCompletion: { _ in
    print("finished")
}, receiveValue: { value in
    print(value)
})

1. We create a publisher Just that sends a single value and then terminates (sends the completion event). Combine has several integrated publishers, including Just.
2. We connect to a Subscriber.

Will print:

1
finished

After sending 1, the publisher automatically terminates. We don't have to handle any errors, since Just cannot fail.

Scheduler

Scheduler is the synchronization mechanism of the Combine framework, which defines the context of where and when work is done. Combine does not work directly with threads. Instead, it allows publishers to operate on specific schedulers. The where means current run loop, dispatch queue or operation queue. The when means virtual time, according to the internal clock of the scheduler. Work performed by a scheduler will adhere only to the scheduler, which may not correspond to real system time.

# Scheduler types

Combine provides different types of schedulers and they all conform to the Scheduler protocol:

• DispatchQueue. It performs work in a specific dispatch queue: serial, concurrent, main, and global. Typically you use serial and global queues for background work, and the main queue for UI-related work. As of
• OperationQueue. Performs work in a specific operation queue. Similar to dispatch queues, it uses OperationQueue.main for UI work and other queues for background work. According to this thread on the Swift forum, it is not recommended to use operation queues with maxConcurrentOperations greater than 1.
• RunLoop. Performs work in the specific runloop.
• ImmediateScheduler. Perform synchronous actions immediately. It will terminate the application with a fatal error if you try to run a postponed job with this scheduler.

Even if you don't specify a scheduler, Combine provides you with the default scheduler. The scheduler uses the same thread, where the element was generated. Let's say if you send the element from the background thread, it receives it in the same background thread.

# Switch between schedulers

Operations that consume resources are usually processed in the background so that the user interface does not freeze. Its result is handled in the main thread. The way Combine does this is by changing the schedulers. It is achieved with the help of two methods: subscribe(on:) and receive(on:). The receive method changes a scheduler for all publishers that come after it. The subscribe method changes the scheduler that is used to perform subscribe, unsubscribe, and request operations. The string will remain in that scheduler all the time, unless specified somewhere in receive. which affects the timing of subscription.

Just(1)
   .subscribe(on: DispatchQueue.global())
   .map { _ in print(Thread.isMainThread) }
   .sink { print(Thread.isMainThread) }

It will print:

false
false

All operations happen in the DispatchQueue.global() scheduler.

Subject

The Subject is a special type of publisher that can insert values, passed from outside, into the stream. The Subject interface provides three different ways to submit elements:

public protocol Subject : AnyObject, Publisher {
    func send(_ value: Self.Output)
    func send(completion: Subscribers.Completion<Self.Failure>)
    func send(subscription: Subscription)
}

Combine has two built-in subjects: PassthroughSubject and CurrentValueSubject.

// 1
let subject = PassthroughSubject<String, Never>()
// 2
subject.sink(receiveCompletion: { _ in
    print("finished")
}, receiveValue: { value in
    print(value)
})
// 3
subject.send("Hello,")
subject.send("World!")
subject.send(completion: .finished) // 4

1. We create a PassthoughSubject. We set failure type to Never to indicate that it always ends successfully.
2. We subscribe to the subject (remember, it is a publisher).
3. We send two values ​​to the stream and then it ends.

The output will be:

Hello,
World!
finished

CurrentValueSubject starts with an initial value and publishes all its changes. You can return its current value through the value property.

// 1
let subject = CurrentValueSubject<Int, Never>(1)
// 2
print(subject.value)
// 3
subject.send(2)
print(subject.value)
// 4
subject.sink { value in
    print(value)
}

1. We create a subject with initial value 1.
2. We access the current value.
3. We update the current value to 2.
4. We subscribe to the subject.

And we have the output:

1
2
2

How to combine a Publisher and a Subscriber

A connection between a publisher and a subscriber is called a subscription. The steps of this connection define a publisher-subscriber life cycle.

let subject = PassthroughSubject<Int, Never>()

let token = subject
    .print()
    .sink(receiveValue: { print("received by subscriber: \($0)") })

subject.send(1)

Note the print(_:to:) operator in this code. We print all log messages for all publish events to the console, which can already tell us a lot about the lifecycle.

This is what is printed to the console:

1. receive subscription: (PassthroughSubject)
2. request unlimited
3. receive value: (1)
4. received by subscriber: 1
5. receive cancel

This gives us a clue about the publisher-subscriber life cycle. Let's examine the life cycle of this connection:

1. A subscriber connects to a Publisher by calling subscribe<S>(S).
2. The publisher creates a subscription by calling receive<S>(subscriber: S) on itself.
3. The publisher acknowledges the subscription request. Call receive(subscription:) on the subscriber.
4. The subscriber requests a quantity of items they wish to receive. Call request(:) on the subscription and pass Demand as a parameter. Demand defines how many objects a publisher can send to a subscriber via subscription. In our case, the demand is unlimited.
5. The publisher sends values ​​by calling receive(_:) of subscriber. This method returns an instance of Demand, which shows how many more items the subscriber expects to receive. The subscriber can only increase the demand or leave it the same, but cannot reduce it.
6. The subscription ends with one of these results:

• Cancelled: this can happen automatically when the subscriber is released, shown in the example above. Another way is to manually cancel with token.cancel()(token is also a subscriber that conforms to the AnyCancellable protocol).
• ***Finish:***end with success
• ***Fail:***fail with the error.

Combine Operators

Operators are special methods that are inside publishers and return another publisher. This allows them to be applied one after another, creating a chain. Each operator transforms the publisher, returned by the previous operator. At the beginning of each chain there must be a publisher. Then the operators can be applied in turn. Each operator receives the publisher created by the previous operator in the chain. We refer to their relative order as upstream and downstream, that is, the immediately preceding and following operator.

Let's see how we can chain operators when handling an HTTP URL request with SwiftUI's Combine framework.

func send<T: Decodable>(_ request: URLRequest, _ decoder: JSONDecoder = JSONDecoder()) -> AnyPublisher<T, APIError> {
        return URLSession.shared
    //1     .dataTaskPublisher(for: request)
    //2     .mapError{ APIError.serverError(code: $0.errorCode, message: $0.localizedDescription) }
    //3     .map { $0.data }
    //4     .decode(type: T.self, decoder: decoder)
    //5     .print()
    //6     .mapError { _ in APIError.decodingError }
    //7     .receive(on: DispatchQueue.main)
    //8     .eraseToAnyPublisher()
    }

1. combine is well integrated with iOS frameworks and SDK. This allows us to use a built-in publisher to handle URLSession data tasks.
2. mapError allows us to transform an internal error into a more descriptive error of our own
3. passes the response data. We use the map operator (_:), which transforms the ascending value of (data: Data, response: URLResponse) to Data.
4. Decode the content of the response with JSONDecoder.
5. we print the content to the console
6. we transform the decoding error if we have it
7. Specifies a scheduler that will receive the publisher elements
8. We remove the publisher type and transform it to AnyPublisher. This way of deleting the type preserves the abstraction of the API, and in this way we can use it in different modules.

Conclusions

Combine is best suited when you want to configure something that reacts to a flow of values. User interfaces fit this pattern very well. But Combine is not limited to user interfaces. Any sequence of asynchronous operations can be effective, especially when the results of each step flow to the next step. An example of this could be a series of network service requests, followed by decoding the results.

Applying these concepts to a strongly typed language like Swift is part of what Apple has created in Combine. Combine extends reactive functional programming by incorporating the concept of backpressure. Backpressure is the idea that the subscriber must control how much information they get at once and need to process. This leads to efficient operation with the added notion that the volume of data processed through a stream is controllable and cancelable.

Combine is used by many Apple frameworks. SwiftUI is the obvious example that gets the most attention, with both subscriber and publisher elements. RealityKit also has publishers that can be used to react to events. And Foundation has a number of Combine-specific additions, including NotificationCenter, URLSession, and Timer as publishers. In fact, any asynchronous API can take advantage of the Combine framework. It is inevitable that as time goes by and as SwiftUI progresses, Combine will be a fundamental part of every Apple platform.

References

1. Apple WWDC Combine Videos
2. Apple Documentation
3. SwiftUI Notes