A publisher temperature primer
29 Apr 2020Here’s a fun1 Combine challenge. What does the following snippet output?
Somewhat surprisingly, the value event for 4 is missing.
This is a subtle gotcha when subscribing to publishers—they implicitly have a “temperature.”
When receiving a subscriber, an upstream publisher hands it a Subscription to work with. The subscription then brokers demand through Subscription.request(_:) and cancellation by way of inheriting from Cancellable. But, that still begs the question, should a publisher “restart” if it receives a new subscriber, even after a completion event? What if the same subscriber attaches twice? Do we doubly send events downstream?
Restarting is helpful when a publisher backs one-off work. Think network requests or logging analytics events. On the other hand, what does it mean for a publisher backing button taps to “restart?”
That’s what temperature delineates.
A cold publisher is one that starts work only when a subscriber is received. It’s a declarative analog to Swift’s lazy keyword. A hot publisher is one that may send events at any time, regardless of the subscriber count. And Subject conformances hang out in the warmer climate.
So, let’s return to that snippet.
Publisher.retry “attempts to recreate a failed subscription with the upstream publisher.” Let’s perform the second subscription attempt manually to investigate further. We’ll need to lean on Publisher.subscribe(_:) (the Subscriber-accepting, not Subject-based overload), but, we had originally used lowercase-s sink and called it a day. The method uses a Subscribers.Sink instance underneath the hood, so we can run with that.
We got two failure events‽ Wat.
No, wait—that makes sense. Publisher.retry suppresses the first error, tries again after the subject becomes inert post-.failure(.anError) and then sends the lone error downstream.
Manually re-subscribing reveals Subject’s warmer temperature. A new subscriber coming around didn’t “restart” it and allow the 4 to come through. Instead, it’s inactive after the first completion, and per the Publisher contract.
The choice of “inactive” there is specific. We can peak into Combine and see that they use the same phrasing.
Subjects keep track of lifecycle state and, if a subscriber shows up after a terminal event, it’s simply told that. Relatedly, if a subscriber is attached to a CurrentValueSubject post-completion, it also misses out on the last value—ReplaySubject fills that gap.
Time to hang with a cooler publisher, URLSession.dataTaskPublisher(for:).
It overlays URLSession with a coating of Combine. Now, It’d be weird if we constructed a URLSession.DataTaskPublisher instance (the type returned from the method) and it went off and requested resources without anyone listening. In fact, doing so might put us in scenarios where we start network traffic before preconditions like authentication are met.
To show how cool this publisher is, let’s tee it up to fail once then succeed on the second attempt.
Close readers might’ve noticed I substituted AnySubscriber for Sink compared to our subject double-subscribing. The reason is two-fold. First, sink (the method) uses a Sink under the hood. Second (and after some investigation), Sink maintains a sort of “subscription status” that cancels any subsequent subscriptions after a first is received.
The folks behind OpenCombine noticed this, too.
Concretely, notice how the output changes when we swap back in Sink (and noting that it isn’t quite synonymous with AnySubscriber since it requests unlimited demand).
Which then raises the question, why did the subject example replay that second failure when coldPublisher didn’t? That’s because warm publishers generally keep track of lifecycle state. If a long-living subject hits a completion event, it knows it can’t further emit values to subscribers, so it hangs onto the completion for future reference. In fact, the dump(subject) we did before revealed that Combine does just that, with an non-public, optional completion property.
And our erased DataTaskPublisher doesn’t keep internal state. We can check this by subscribing two distinct Sinks (with some delay to prevent overlap).
Using different Sinks restarts coldPublisher. So, how do individual Sink instances keep state? To answer this, I tried the following for some clues:
No dump or mirrored children? That’s odd. The type’s listed conformances hint at why:
CustomReflectable, huh? I guess—and if you know the reason, please let me know (!)—Apple is trying to hide their implementation here by returning an empty mirroring? We’d at least expect the @escaping receiveCompletion and receiveValue closures to be reflected.
And again, OpenCombine caught this, returning an EmptyCollection in their Subscribers.Sink.customMirror implementation (seriously, bravo to Sergej and the project’s contributors).
Despite this, we can verify that Sink keeps track of lifecycle state by subscribing a single instance, twice, to a publisher that finishes the first subscription and see how it affects the second.
Aha. The second received subscription is immediately cancelled after it received a finished event from the first.
…that was a winding detour—let’s drive back and recap where we’re at.
Publishers come in two temperatures: hot and cold. But, it’s sometimes tricky to elicit a cold publisher’s “coldness” because Subscribers.Sink keeps lifecycle state that cancels subsequent subscriptions.
That brings us to another juncture, how do we tell publisher types apart and switch between them?
Checking temperatures
I wish Combine had a better story here because, in short, there isn’t a great way to statically check if a publisher is hot or cold.
Some of Combine’s prior art made the distinction at the type level with Signal and SignalProducer, respectively.
This does come with an ergonomic cost, since you then have to juggle output, failure, and temperature types when building pipelines. Still, better to check at compile time instead of running a fever at runtime or in production (sorry, that pun was bad).
Another, more manual approach is to trace any side effects performed by cold publishers. That is, open up Proxyman, turn on analytics logging, and the like to make sure effects are being run when and how often you expect.
If they aren’t, then it’s time to heat up or cool things down.
Heating and cooling
Hot ⇒ cold
Deferred { Future<_, _> { promise in /* … */ } } is a dance you’ll see often in everyday Combine. It cools down an eagerly-evaluated-by-default Future into one that kicks off its attemptToFulfill closure only when a subscriber is received.
(But even after deferring, there’s a tangential gotcha. Deferred is a struct, which means every call site gets its own copy and in turn, creates its own Future. For reference semantics, you’ll want to tack on a share operator.)
If you have a subject in hand, it’s not so much about changing temperature—since there’s no “work” backing them—and instead about caching values for future subscribers. CurrentValueSubject will do just that with the latest value and before any completion events are received. To replay more than one value, even after completion, you’ll need to reach for a ReplaySubject.
Cold ⇒ hot
Warming up a cold publisher is either a share or multicast call away. The former simply converts from value to reference semantics, while the latter can be used for more control on how events are projected down to subscribers. In fact, it can even be used to implement both share and share(replay:).
My entry on multicasting walks through this.
⬦
Back to our original quiz with what we learned—and the friends we made—along the way.
4 isn’t received since PassthroughSubject is warm and retry’s attempt to create another subscription is met with a replayed .failure(.anError) event that then ends the sequence.
A whole lot of detail packed into 25 lines, eh?
■
Related reading and footnotes
⇒ The “Hot and cold publishers” section of Part II of Matt Gallagher’s three-part Combine series.
⇒ OpenCombine’s implementation of Subscribers.Sink.
⇒ Understanding Combine’s section on Future and Deferred.
-
For some hyper-specific definition of “fun.” ↩