Stopping a Pipeline

We now know how to construct a pipeline and start it going: in general, simply constructing the entire pipeline, including the final subscriber, does start it going (although there are some types of publisher where it takes a little more than that). But how does a pipeline, or part of a pipeline, come to a stop?

There are three chief ways: cancellation, completion, and error. I’ll talk about cancellation and completion here, and I’ll discuss error in the next section.

Cancellation

The way that a publisher and a subscriber break their connection in good order is through cancellation: the subscriber cancels its subscription to the publisher. (Once again, the actual medium of communication is the subscription object that they share.) This causes the subscriber to be unsubscribed, and causes the publisher to stop publishing.

If the publisher is an operator, meaning that it has an upstream publisher, it turns around and uses the same mechanism, sending a cancel message to that publisher. And so the cancel message percolates up the pipeline, until it reaches the ultimate publisher, which stops publishing.

It turns out that you are allowed to participate directly in this mechanism. You can turn off a pipeline manually by telling the subscriber at the end of the pipeline to cancel, thus causing the cancel message to percolate up through the entire pipeline all the way to the ultimate publisher, and terminating the whole pipeline, permanently, in good order. You can do that, provided the subscriber adopts the Cancellable protocol, which means it has a cancel method. A Sink object does adopt the Cancellable protocol.

However, we have no direct access to any Sink object. As I mentioned earlier, we are storing our Sink object in a Set called storage by calling store(in:) on it. How does that work? Well, it turns out that the .sink method does not merely produce a Sink object; it also wraps that object in an instance of a type-erasing wrapper class called AnyCancellable. The store(in:) method is an AnyCancellable method; it stores the AnyCancellable object in the collection we designate. Here, that’s a Set whose elements are typed (appropriately enough) as AnyCancellable.

Well, AnyCancellable also has a cancel method! So we can tell an AnyCancellable to cancel, it passes that message along to its wrapped Cancellable subscriber, and the pipeline is terminated and the operators are cancelled, in good order, all the way up to the publisher, which stops publishing.

We’re keeping our AnyCancellable wrapper inside the self.storage Set. So all we have to do is find it and tell it to cancel. If we have only one pipeline, it will be the first element of the Set:

self.storage.first?.cancel()

(If we have more than one pipeline, we will need some additional reference to this pipeline in order to distinguish it and tell it to cancel. But there’s no need to go into that right now.)

Another way to do the same thing would be to remove the AnyCancellable wrapper from the Set:

self.storage.removeFirst()

The reason that works is that an AnyCancellable object, which is a reference type (a class), automatically cancels itself when it goes out of existence — which is just what happens when we remove it from its Set and nothing is retaining it any longer.

That, indeed, is one of the main reasons why store(in:) is a good thing: it means that when self (here, a view controller) goes out of existence, any pipelines it is retaining in a property such as our self.storage are terminated in good order. When our view controller goes out of existence, its properties will be released, the AnyCancellables inside our storage Set will be released, the AnyCancellables will all get cancel messages automatically, they’ll pass those along to their wrapped subscribers, the messages will percolate up the respective pipelines, and all the corresponding publishers will stop publishing and all the objects in the pipeline will be released.

Return, for example, to our notification center publisher and its pipeline:

NotificationCenter.default.publisher(
    for: UIWindowScene.didEnterBackgroundNotification)
    .sink { _ in
        print("we're going into the background!")
    }
    .store(in:&self.storage)

We might configure that pipeline in our view controller’s viewDidLoad. And that is all we have to do, because when our view controller goes out of existence, the pipeline will be cancelled automatically and the notification center publisher will stop publishing.

Completion

Some publishers can produce values from time to time forever; they just keep publishing until they are told to stop (cancellation). Our notification center publisher is a case in point:

NotificationCenter.default.publisher(
    for: UIWindowScene.didEnterBackgroundNotification)
    .sink { _ in
        print("we're going into the background!")
    }
    .store(in:&self.storage)

That publisher stands ready to notify us, by producing a value, any time our window scene goes into the background, as many times as needed, forever — meaning as long as the app runs and the pipeline exists.

But some publishers provide only a finite number of values and then stop; their job is done. The data task publisher from the previous section is a case in point:

URLSession.shared.dataTaskPublisher(for: url)
    .compactMap { UIImage(data:$0.data) }
    .receive(on: DispatchQueue.main)
    .sink(receiveCompletion: {_ in}) {
        self.iv.image = $0
    }
    .store(in:&self.storage)

That publisher, by default, is going to go out on the Internet once and try to download the data from the URL it is given. The publisher either succeeds or fails in providing an image, but either way, having made the attempt, its job is done.

When a publisher has done its job, it needs to let its downstream object know that no more values are to be expected. It does that by passing a completion message to the downstream object. This message is actually an enum; it is the .finished case of the Subscribers.Completion enum. (The medium of communication is once more the subscription object that I mentioned earlier, but that’s just an implementation detail you usually won’t need to worry about). If this object is an operator, it usually finishes doing whatever it’s doing and passes a completion message to its downstream object — and so on, down the pipeline.

Typically, in this way, the completion message reaches the final subscriber, and the pipeline is now terminated in good order. The effect is the same as cancellation. A subsequent attempt to cancel the pipeline will do nothing; the pipeline has already been torn down.