Intermediate Combine

When a Subscriber initially calls subscription.request(_:), it specifies the maximum number of values it is willing to receive.

If you take a look at the protocol description for Subscriber, you’ll see that the receieve method returns a Demand

func receive(_ input: Self.Input) -> Subscribers.Demand  //ON SLIDE

In receive(_:), you can adjust that max number of values it is willing to receive each time a new value is received. This allows you to dynamically manage backpressure.

One thing to note however is that the adjustment is additive - the new max is added to the current max. Also the new max must be positive, meaning you can increase the value of max, but never decrease it. Let’s take a look at this in an example.

Start an example block, and within it create an IntSubscriber class which adopts Subscriber

example(of: "Dynamically adjusting Demand") {
  final class IntSubscriber: Subscriber {

Make type aliases for the Input and Failure types

    typealias Input = Int
    typealias Failure = Never

The make the 3variants of the receive function in the Subscriber protocol. In the first one, call request on the passed in subscription, setting the max to 2.

    func receive(subscription: Subscription) {
      subscription.request(.max(2))
    }

In the second receive function, return a new demand value based on the input value. Remember this max value is additive.

    func receive(_ input: Int) -> Subscribers.Demand {
      print("Received value", input)
      
      switch input {
      case 1:
        return .max(2) // 1
      case 3:
        return .max(1) // 2
      default:
        return .none // 3
      }
    }

In the final receive function, which takes in the completion, simply output a message to console.

    func receive(completion: Subscribers.Completion<Never>) {
      print("Received completion", completion)
    }
  }

Initialize an IntSubscriber, create a PassthroughSubject that handles Integer data and Never returns errors, and subscribe to that subject with the subscriber.

  let subscriber = IntSubscriber()
  
  let subject = PassthroughSubject<Int, Never>()
  
  subject.subscribe(subscriber)

Send values to the subject to push them through the pipeline.

  subject.send(1)
  subject.send(2)
  subject.send(3)
  subject.send(4)
  subject.send(5)
  subject.send(6)
}

Run this in a playground, and you’ll see that the first 5 values are emitted, but the sixth is not. To see why, look at the value of max in the second receive function.

When the input is 1, the new max is 4 (original max of 2 + new max of 2). When the input is 3, the new max is 5 (previous 4 + new 1). For all other values, the max remains what it was, which for the last entry is 5.