How Combine works: Operators

In my previous post, I explained subscriptions and I want to focus now on operators.

As opposed to Subscription and Publisher operators don't have an interface of their own, they really are just publishers that subscribe to upstream publishers, receive the data from there, do something with it, and then send data to downstream publishers:

   Just(5)
    .map {$0 + 1}
    .map { $0 + 2 }
    .sink { print("we got \($0)") }

In this example, Just is our most upstream publisher, the two map publishers are what we call operators; the first one subscribes to Just adds 1 to that value and sends it downstream where another map once again, having subscribed to its upstream map publisher, adds 2 and sends the value downstream, this time to our end client, the sink subscriber.

So first, notice map as opposed to Just is not a type, but a method. Well, sort of.

To make it more convenient so that you don't have to creat a Map publisher yourself, Apple wrote map as an extension function that any publisher can have access to, something like:

extension Publisher {
    public func map<T>(_ transform: @escaping (Self.Output) -> T) -> Publishers.Map<Self, T>
}

Ok, so if Map returned by map is itself a Publisher that explains how downstream sink subscriber can subscribe to Map, right? It's a publisher, you can subscribe to it.

However, for me, the big question was, how did Map as a publisher, subscribe to its own upstream publisher?

Let's focus on this:

Just(5)
   .map {$0 + 1}

If map is only a publisher how does it subscribe to Just? Since it doesn't implement the Subscriber protocol?

The answer how this almost magic chain works is AnySubscriber. Quoting Apple documentation:

Use an AnySubscriber to wrap an existing subscriber whose details you don’t want to expose

To wrap an existing subscriber! Yes, it's also to hide implementation details or so, but what's most important is the following initializer AnySubscriber provides:

init(receiveSubscription: ((Subscription) -> Void)?, receiveValue: ((Input) -> Subscribers.Demand)?, receiveCompletion: ((Subscribers.Completion<Failure>) -> Void)?)

Creates a type-erasing subscriber that executes the provided closures.

In essence, what Map does is, once it receive its subscriber from downstream it wraps it in AnySubscriber and adds its logic, so that it can capture values from upstream and change them, when they are provided.

Imagine Maps receive<S>(subscriber: S) implementations as such:

    public func receive<S>(subscriber: S) where Output == S.Input, S : Subscriber, Upstream.Failure == S.Failure {
        let newSubscriber = AnySubscriber(receiveSubscription: { subscriber.receive(subscription: $0)},
                      receiveValue: { val in
                                         let newVal = // ... apply closure; add 1, in our caase
                                         subscriber.receive(newVal) },
                      receiveCompletion: { subscriber.receive(completion: $0) } )
        upstream.subscribe(newSubscriber)
    }

Map is wrapping the initial subscriber and then passing it forward, this allows it to capture data and transform it appropriately.