How Combine works: Subscriptions

In my previous post, I showed a basic, rudimentary Publisher and introduced elementary notions of Combine.framework.

If you want to find out how operators work, you can checkout my final post for the series.

Now, we will focus on Subscriptions. Normally, any decent publisher only starts sending values (well, often they don't do it directly, the subscription does it for them), only when it's requested. It honors that request, and after each time it sends a value, it checks to see if:

• additional values are demanded or if none are demanded any longer
• if the subscription has been canceled

That's what we will do today. For that we will write a Subscription class, I say class because as opposed to the publisher, this one gets passed around, so it makes sense to use a reference type.

The protocol Subscription:

protocol Subscription : Cancellable, CustomCombineIdentifierConvertible 

demands of us:

func request(Subscribers.Demand)
func cancel()

Notice a subscription is cancelable, so it can be canceled at any time. This is why subscriptions need be stored somewhere once the subscriber has received it form the publisher. Because losing that reference, when it goes out of scope, Swift will automatically call cancel for us.

Also notice the request method. It gets called by the subscriber whenever he feels like. A subscriber should not receive values until he calls the request method with an initial demand. Subscribers.Demand can be:

• none (.none) don't give me anything
• unlimited (.unlimited) give me infinite values
• max(Int?) give me a finite amount of values

That's when we can start sending values to our subscriber. And we should also honor the demand, not to go above requested values. Also, demand can be called multiple times by the subscriber, it's also our decision if we can honor all that those demands or not. But it's something we should keep in mind.

Here's our custom Subscriber:

 extension Until {
     public class Sub: Subscription {

        private var subscriber: AnySubscriber<Int,Never>?

        private var demand: Subscribers.Demand = .none


         private let endVal: Int


         func request(_ demand: Subscribers.Demand) {
             self.demand = demand
             // only send if there is demand
             if demand != .none {
                 startSending()
             }
        }

         public init<S>(_ subscriber: S, _ endVal: Int) where S : Subscriber, Never == S.Failure, Int == S.Input {
            self.subscriber = AnySubscriber(subscriber)
            self.endVal = endVal
        }


         private func startSending() {
             for value in (0...endVal) {
                 // make sure subscriber has not canceled
                 // or he's demand is lower that what the publisher told us
                 guard let subscriber,
                       self.demand <= value else {
                     return
                 }
              // after every receive call, the subscriber can ask for more, or less...
                let additonalDemanand = subscriber.receive(value)
                demand += additonalDemanand
            }
           // when done send the trigger event
             subscriber?.receive(completion: .finished)
        }

       // if the subscriber cancels, we honor that and terminate our reference to him
         func cancel() {
            subscriber = nil
        }

    }
}

In our publisher from the previous post, we can update the receive logic to:

    public func receive<S>(subscriber: S) where S : Subscriber, Never == S.Failure, Int == S.Input {
        let subscription = Until.Sub(subscriber, endValue)
        subscriber.receive(subscription: subscription)
    }