18. Custom Publishers & Handling Backpressure

Written by Florent Pillet

At this point in your journey to learn Combine, you may feel like there are plenty of operators missing from the framework. This may be particularly true if you have experience with other reactive frameworks, which typically provide a rich ecosystem of operators, both built-in and third-party. Combine allows you to create your own publishers. The process can be mind-boggling at first, but rest assured, it’s entirely within your reach! This chapter will show you how.

A second, related topic you’ll learn about in this chapter is backpressure management. This will require some explanation: What is this backpressure thing? Is that some kind of back pain induced by too much leaning over your chair, scrutinizing Combine code? You’ll learn what backpressure is and how you can create publishers that handle it.

Creating your own publishers

The complexity of implementing your own publishers varies from “easy” to “pretty involved.” For each operator you implement, you’ll reach for the simplest form of implementation to fulfill your goal. In this chapter, you’ll look at three different ways of crafting your own publishers:

• Using a simple extension method in the Publisher namespace.
• Implementing a type in the Publishers namespace with a Subscription that produces values.
• Same as above, but with a subscription that transforms values from an upstream publisher.

Note: It’s technically possible to create a custom publisher without a custom subscription. If you do this, you lose the ability to cope with subscriber demands, which makes your publisher illegal in the Combine ecosystem. Early cancellation can also become an issue. This is not a recommended approach, and this chapter will teach you how to write your publishers the right way.

Publishers as extension methods

Your first task is to implement a simple operator just by reusing existing operators. This is as simple as you can get.

To do it, you’ll add a new unwrap() operator, which unwraps optional values and ignores their nil values. It’s going to be a very simple exercise, as you can reuse the existing compactMap(_:) operator, which does just that, although it requires you to provide a closure.

Using your new unwrap() operator will make your code easier to read, and it will make what you’re doing very clear. The reader won’t even have to look at the contents of a closure.

You’ll add your operator in the Publisher namespace, as you do with all other operators.

Open the starter playground for this chapter, which can be found in projects/Starter.playground and open its Unwrap operator page from the Project Navigator.

Then, add the following code:

extension Publisher {
  // 1
  func unwrap<T>() -> Publishers.CompactMap<Self, T> where Output == Optional<T> {
    // 2
    compactMap { $0 }
  }
}

1. The most complicated part of writing a custom operator as a method is the signature. Read on for a detailed description.
2. Implementation is trivial: Simply use compactMap(_:) on self!

The method signature can be mind-boggling to craft. Break it down to see how it works:

func unwrap<T>()

Your first step is to make the operator generic, as its Output is the type the upstream publisher‘s optional type wraps.

-> Publishers.CompactMap<Self, T>

The implementation uses a single compactMap(_:), so the return type derives from this. If you look at Publishers.CompactMap, you see it’s a generic type: public struct CompactMap<Upstream, Output>. When implementing your custom operator, Upstream is Self (the publisher you’re extending) and Output is the wrapped type.

where Output == Optional<T> {

Finally, you constrain your operator to Optional types. You conveniently write it to match the wrapped type T with your method’s generic type… et voilà!

Note: When developing more complex operators as methods, such as when using a chain of operators, the signature can quickly become very complicated. A good technique is to make your operators return an AnyPublisher<OutputType, FailureType>. In the method, you’ll return a publisher that ends with eraseToAnyPublisher() to type-erase the signature.

Testing your custom operator

Now you can test your new operator. Add this code below the extension:

let values: [Int?] = [1, 2, nil, 3, nil, 4]

values.publisher
  .unwrap()
  .sink {
    print("Received value: \($0)")
  }

Run the playground and, as expected, only the non-nil values are printed out to the debug console:

Received value: 1
Received value: 2
Received value: 3
Received value: 4

Now that you’ve learned about making simple operator methods, it’s time to dive into richer, more complicated publishers. You can group publishers like so:

• Publishers that act as “producers” and directly produce values themselves.
• Publishers that act as “transformers,” transforming values produced by upstream publishers.

In this chapter, you’ll learn how to use both, but you first need to understand the details of what happens when you subscribe to a publisher.

The subscription mechanism

Subscriptions are the unsung heroes of Combine: While you see publishers everywhere, they are mostly inanimate entities. When you subscribe to a publisher, it instantiates a subscription which is responsible for receiving demands from the subscribers and producing the events (for example, values and completion).

Here are the details of the lifecycle of a subscription:

1. A subscriber subscribes to the publisher.
2. The publisher creates a Subscription then hands it over to the subscriber (calling receive(subscription:)).
3. The subscriber requests values from the subscription by sending it the number of values it wants (calling the subscription’s request(_:) method).
4. The subscription begins the work and starts emitting values. It sends them one by one to the subscriber (calling the subscriber’s receive(_:) method).
5. Upon receiving a value, the subscriber returns a new Subscribers.Demand, which adds to the previous total demand.
6. The subscription keeps sending values until the number of values sent reaches the total requested number.

If the subscription has sent as many values as the subscriber has requested, it should wait for a new demand request before sending more. You can bypass this mechanism and keep sending values, but that breaks the contract between the subscriber and the subscription and can cause undefined behavior in your publisher tree based on Apple’s definition.

Finally, if there is an error or the subscription’s values source completes, the subscription calls the subscriber’s receive(completion:) method.

Publishers emitting values

In Chapter 11, “Timers,” you learned about Timer.publish() but found that using Dispatch Queues for timers was somewhat uneasy. Why not develop your own timer based on Dispatch’s DispatchSourceTimer?

You’re going to do just that, checking out the details of the Subscription mechanism while you do.

To get started, open the DispatchTimer publisher page of the playground.

You’ll start by defining a configuration structure, which will make it easy to share the timer configuration between the subscriber and its subscription. Add this code to the playground:

struct DispatchTimerConfiguration {
  // 1
  let queue: DispatchQueue?
  // 2
  let interval: DispatchTimeInterval
  // 3
  let leeway: DispatchTimeInterval
  // 4
  let times: Subscribers.Demand
}

If you’ve ever used DispatchSourceTimer, some of these properties should look familiar to you:

1. You want your timer to be able to fire on a certain queue, but you also want to make the queue optional if you don’t care. In this case, the timer will fire on a queue of its choice.
2. The interval at which the timer fires, starting from the subscription time.
3. The leeway, which is the maximum amount of time after the deadline that the system may delay the delivery of the timer event.
4. The number of timer events you want to receive. Since you’re making your own timer, make it flexible and able to deliver a limited number of events before completing!

Adding the DispatchTimer publisher

You can now start creating your DispatchTimer publisher. It’s going to be straightforward because all the work occurs inside the subscription!

Add this code below your configuration:

extension Publishers {
  struct DispatchTimer: Publisher {
    // 5
    typealias Output = DispatchTime
    typealias Failure = Never
    // 6
    let configuration: DispatchTimerConfiguration
    
    init(configuration: DispatchTimerConfiguration) {
      self.configuration = configuration
    }
  }
}

5. Your timer emits the current time as a DispatchTime value. Of course, it never fails, so the publisher’s Failure type is Never.
6. Keeps a copy of the given configuration. You don’t use it right now, but you’ll need it when you receive a subscriber.

Note: You’ll start seeing compiler errors as you write your code. Rest assured that you’ll remedy these by the time you’re done implementing the requirements.

Now, implement the Publisher protocol’s required receive(subscriber:) method by adding this code to the DispatchTimer definition, below your initializer:

// 7
func receive<S: Subscriber>(subscriber: S)
  where Failure == S.Failure,
        Output == S.Input {
  // 8
  let subscription = DispatchTimerSubscription(
    subscriber: subscriber,
    configuration: configuration
  )
  // 9
  subscriber.receive(subscription: subscription)
}

7. The function is a generic one; it needs a compile-time specialization to match the subscriber type.
8. The bulk of the action will happen inside the DispatchTimerSubscription that you’re going to define in a short while.
9. As you learned in Chapter 2, “Publishers & Subscribers,” a subscriber receives a Subscription, which it can then send requests for values to.

That’s really all there is to the publisher! The real work will happen inside the subscription itself.

Building your subscription

The subscription’s role is to:

• Accept the initial demand from the subscriber.
• Generate timer events on demand.
• Add to the demand count every time the subscriber receives a value and returns a demand.
• Make sure it doesn’t deliver more values than requested in the configuration.

This may sound like a lot of code, but it’s not that complicated!

Start defining the subscription below the extension on Publishers:

private final class DispatchTimerSubscription
  <S: Subscriber>: Subscription where S.Input == DispatchTime {
}

The signature itself gives a lot of information:

• This subscription is not visible externally, only through the Subscription protocol, so you make it private.
• It’s a class because you want to pass it by reference. The subscriber may then add it to a Cancellable collection, but also keep it around and call cancel() independently.
• It caters to subscribers whose Input value type is DispatchTime, which is what this subscription emits.

Adding required properties to your subscription

Now add these properties to the subscription class’ definition:

// 10
let configuration: DispatchTimerConfiguration
// 11
var times: Subscribers.Demand
// 12
var requested: Subscribers.Demand = .none
// 13
var source: DispatchSourceTimer? = nil
// 14
var subscriber: S?

This code contains:

10. The configuration that the subscriber passed.
11. The maximum number of times the timer will fire, which you copied from the configuration. You’ll use it as a counter that you decrement every time you send a value.
12. The current demand; e.g., the number of values the subscriber requested — you decrement it every time you send a value.
13. The internal DispatchSourceTimer that will generate the timer events.
14. The subscriber. This makes it clear that the subscription is responsible for retaining the subscriber for as long as it doesn’t complete, fail or cancel.

Note: This last point is crucial to understand the ownership mechanism in Combine. A subscription is the link between a subscriber and a publisher. It keeps the subscriber — for example, an object holding closures, like AnySubscriber or sink — around for as long as necessary. This explains why, if you don’t hold on to a subscription, your subscriber never seems to receive values: Everything stops as soon as the subscription is deallocated. Internal implementation may of course vary according to the specifics of the publisher you are coding.

Initializing and canceling your subscription

Now, add an initializer to your DispatchTimerSubscription definition:

init(subscriber: S,
     configuration: DispatchTimerConfiguration) {
  self.configuration = configuration
  self.subscriber = subscriber
  self.times = configuration.times
}

This is pretty straightforward. The initializer sets times to the maximum number of times the publisher should receive timer events, as the configuration specifies. Every time the publisher emits an event, this counter decrements. When it reaches zero, the timer completes with a finished event.

Now, implement cancel(), a required method that a Subscription must provide:

func cancel() {
  source = nil
  subscriber = nil
}

Setting DispatchSourceTimer to nil is enough to stop it from running. Setting the subscriber property to nil releases it from the subscription’s reach. Don’t forget to do this in your own subscriptions to make sure you don’t retain objects in memory that are no longer needed.

You can now start coding the core of the subscription: request(_:).

Letting your subscription request values

Do you remember what you learned in Chapter 2, “Publishers & Subscribers?” Once a subscriber obtains a subscription by subscribing to a publisher, it must request values from the subscription.

This is where all the magic happens. To implement it, add this method to the class, above the cancel method:

// 15
func request(_ demand: Subscribers.Demand) {
  // 16
  guard times > .none else {
    // 17
    subscriber?.receive(completion: .finished)
    return
  }
}

15. This required method receives demands from the subscriber. Demands are cumulative: They add up to form a total number of values that the subscriber requested.
16. Your first test is to verify whether you’ve already sent enough values to the subscriber, as specified in the configuration. That is, if you’ve sent the maximum number of expected values, independent of the demands your publisher received.
17. If this is the case, you can notify the subscriber that the publisher has finished sending values.

Continue the implementation of this method by adding this code after the guard statement:

// 18
requested += demand

// 19
if source == nil, requested > .none {
  
}

18. Increment the total number of values requested by adding the new demand.
19. Check whether the timer already exists. If not, and if requested values exist, then it’s time to start it.

Configuring your timer

Add this code to the body of this last if conditional:

// 20
let source = DispatchSource.makeTimerSource(queue: configuration.queue)
// 21
source.schedule(deadline: .now() + configuration.interval,
                repeating: configuration.interval,
                leeway: configuration.leeway)

20. Create the DispatchSourceTimer from the queue you configured.
21. Schedule the timer to fire after every configuration.interval seconds.

Once the timer has started, you’ll never stop it, even if you don’t use it to emit events to the subscriber. It will keep running until the subscriber cancels the subscription — or you deallocate the subscription.

You’re now ready to code the core of your timer, which emits events to the subscriber. Still inside the if body, add this code:

// 22
source.setEventHandler { [weak self] in
  // 23
  guard let self = self,
        self.requested > .none else { return }

  // 24
  self.requested -= .max(1)
  self.times -= .max(1)
  // 25
  _ = self.subscriber?.receive(.now())
  // 26
  if self.times == .none {
    self.subscriber?.receive(completion: .finished)
  }
}

22. Set the event handler for your timer. This is a simple closure the timer calls every time it fires. Make sure to keep a weak reference to self or the subscription will never deallocate.
23. Verify that there are currently requested values — the publisher could be paused with no current demand, as you’ll see later in this chapter when you learn about backpressure.
24. Decrement both counters now that you’re going to emit a value.
25. Send a value to the subscriber.
26. If the total number of values to send meets the maximum that the configuration specifies, you can deem the publisher finished and emit a completion event!

Activating your timer

Now that you’ve configured your source timer, store a reference to it and activate it by adding this code after setEventHandler:

self.source = source
source.activate()

That was a lot of steps, and it would be easy to inadvertently misplace some code along the way. This code should have cleared all the errors in the playground. If it hasn’t, you can double-check your work by reviewing the above steps or by comparing your code with the finished version of the playground in projects/Final.playground.

Last step: Add this extension after the entire definition of DispatchTimerSubscription, to define an operator that makes it easy to chain this publisher:

extension Publishers {
  static func timer(queue: DispatchQueue? = nil,
                    interval: DispatchTimeInterval,
                    leeway: DispatchTimeInterval = .nanoseconds(0),
                    times: Subscribers.Demand = .unlimited)
                    -> Publishers.DispatchTimer {
    return Publishers.DispatchTimer(
      configuration: .init(queue: queue,
                           interval: interval,
                           leeway: leeway,
                           times: times)
                      )
  }
}

Testing your timer

You’re now ready to test your new timer!

Most parameters of your new timer operator, except the interval, have a default value to make it easier to use in common use cases. These defaults create a timer that never stops, has minimal leeway and don’t specify which queue it wants to emit values on.

Add this code after the extension to test your timer:

// 27
var logger = TimeLogger(sinceOrigin: true)
// 28
let publisher = Publishers.timer(interval: .seconds(1),
                                 times: .max(6))
// 29
let subscription = publisher.sink { time in
  print("Timer emits: \(time)", to: &logger)
}

27. This playground defines a class, TimeLogger, that’s very similar to the one you learned to create in Chapter 10, “Debugging.” The only difference is this one can display either the time difference between two consecutive values, or the elapsed time since the timer was created. Here, you want to display the time since you started logging.
28. Your timer publisher will fire exactly six times, once every second.
29. Log each value you receive through your TimeLogger.

Run the playground and you‘ll see this nice output — or something similar, since the timing will vary slightly:

+1.02668s: Timer emits: DispatchTime(rawValue: 183177446790083)
+2.02508s: Timer emits: DispatchTime(rawValue: 183178445856469)
+3.02603s: Timer emits: DispatchTime(rawValue: 183179446800230)
+4.02509s: Timer emits: DispatchTime(rawValue: 183180445857620)
+5.02613s: Timer emits: DispatchTime(rawValue: 183181446885030)
+6.02617s: Timer emits: DispatchTime(rawValue: 183182446908654)

There’s a slight offset at setup — and there can also be some added delay coming from Playgrounds — and then the timer fires every second, six times.

You can also test canceling your timer, for example, after a few seconds. Add this code to do so:

DispatchQueue.main.asyncAfter(deadline: .now() + 3.5) {
  subscription.cancel()
}

Run the playground again. This time, you only see three values. It looks like your timer works just fine!

Although it’s barely visible in the Combine API, Subscription does the bulk of the work, as you just discovered.

Enjoy your success. You’ve got another deep dive coming up next!

Publishers transforming values

You’ve made serious progress in building your Combine skills! You can now develop your own operators, even fairly complex ones. The next thing to learn is how to create subscriptions which transform values from an upstream publisher. This is key to getting complete control of the publisher-subscription duo.

In Chapter 9, “Networking,” you learned about how useful sharing a subscription is. When the underlying publisher is performing significant work, like requesting data from the network, you want to share the results with multiple subscribers. However, you want to avoid issuing the same request multiple times to retrieve the same data.

It can also be beneficial to replay the results to future subscribers if you don’t need to perform the work again.

Why not try and implement shareReplay(), which can do exactly what you need? This will be an interesting task! To write this operator, you’ll create a publisher that does the following:

• Subscribes to the upstream publisher upon the first subscriber.
• Replays the last N values to each new subscriber.
• Relays the completion event, if one emitted beforehand.

Beware that this will be far from trivial to implement, but you’ve definitely got this! You’ll take it step by step and, by the end, you’ll have a shareReplay() that you can use in your future Combine-driven projects.

Open the ShareReplay operator page in the playground to get started.

Implementing a ShareReplay operator

To implement shareReplay() you’ll need:

• A type conforming to the Subscription protocol. This is the subscription each subscriber will receive. To make sure you can cope with each subscriber’s demands and cancellations, each one will receive a separate subscription.
• A type conforming to the Publisher protocol. You’ll implement it as a class because all subscribers want to share the same instance.

Start by adding this code to create your subscription class:

// 1
fileprivate final class ShareReplaySubscription<Output, Failure: Error>: Subscription {
  // 2
  let capacity: Int
  // 3
  var subscriber: AnySubscriber<Output,Failure>? = nil
  // 4
  var demand: Subscribers.Demand = .none
  // 5
  var buffer: [Output]
  // 6
  var completion: Subscribers.Completion<Failure>? = nil
}

From the top:

1. You use a generic class, not a struct, to implement the subscription: Both the Publisher and the Subscriber need to access and mutate the subscription.
2. The replay buffer’s maximum capacity will be a constant that you set during initialization.
3. Keeps a reference to the subscriber for the duration of the subscription. Using the type-erased AnySubscriber saves you from fighting the type system. :]
4. Tracks the accumulated demands the publisher receives from the subscriber so that you can deliver exactly the requested number of values.
5. Stores pending values in a buffer until they are either delivered to the subscriber or thrown away.
6. This keeps the potential completion event around, so that it’s ready to deliver to new subscribers as soon as they begin requesting values.

Note: If you feel that it’s unnecessary to keep the completion event around when you’ll just deliver it immediately, rest assured that’s not the case. The subscriber should receive its subscription first, then receive a completion event — if one was previously emitted — as soon as it is ready to accept values. The first request(_:) it makes signals this. The publisher doesn’t know when this request will happen, so it just hands the completion over to the subscription to deliver it at the right time.

Initializing your subscription

Next, add the initializer to the subscription definition:

init<S>(subscriber: S,
        replay: [Output],
        capacity: Int,
        completion: Subscribers.Completion<Failure>?)
        where S: Subscriber,
              Failure == S.Failure,
              Output == S.Input {
  // 7
  self.subscriber = AnySubscriber(subscriber)
  // 8
  self.buffer = replay
  self.capacity = capacity
  self.completion = completion
}

This initializer receives several values from the upstream publisher and sets them on this subscription instance. Specifically, it:

7. Stores a type-erased version of the subscriber.
8. Stores the upstream publisher’s current buffer, maximum capacity and completion event, if emitted.

Sending completion events and outstanding values to the subscriber

You’ll need a method which relays completion events to the subscriber. Add the following to the subscription class to satisfy that need:

private func complete(with completion: Subscribers.Completion<Failure>) {
  // 9
  guard let subscriber = subscriber else { return }
  self.subscriber = nil
  // 10
  self.completion = nil
  self.buffer.removeAll()
  // 11
  subscriber.receive(completion: completion)
}

This private method does the following:

9. Keeps the subscriber around for the duration of the method, but sets it to nil in the class. This defensive action ensures any call the subscriber may wrongly issue upon completion will be ignored.
10. Makes sure that completion is sent only once by also setting it to nil, then empties the buffer.
11. Relays the completion event to the subscriber.

You’ll also need a method that can emit outstanding values to the subscriber. Add this method to emit values as needed:

private func emitAsNeeded() {
  guard let subscriber = subscriber else { return }
  // 12
  while self.demand > .none && !buffer.isEmpty {
    // 13
    self.demand -= .max(1)
    // 14
    let nextDemand = subscriber.receive(buffer.removeFirst())
    // 15
    if nextDemand != .none {
      self.demand += nextDemand
    }
  }
  // 16
  if let completion = completion {
    complete(with: completion)
  }
}

First, this method ensures there is a subscriber. If there is, the method will:

12. Emit values only if it has some in the buffer and there’s an outstanding demand.
13. Decrement the outstanding demand by one.
14. Send the first outstanding value to the subscriber and receive a new demand in return.
15. Add that new demand to the outstanding total demand, but only if it’s not .none. Otherwise, you’ll get a crash, because Combine doesn’t treat Subscribers.Demand.none as zero and adding or subtracting .none will trigger an exception.
16. If a completion event is pending, send it now.

Things are shaping up! Now, implement Subscription’s all-important requirement:

func request(_ demand: Subscribers.Demand) {
  if demand != .none {
    self.demand += demand
  }
  emitAsNeeded()
}

That was an easy one. Remember to check for .none to avoid crashes — and to keep an eye out to see future versions of Combine fix this issue — and then proceed emitting.

Note: calling emitAsNeeded() even if the demand is .none guarantees that you properly relay a completion event that has already occurred.

Canceling your subscription

Canceling the subscription is even easier. Add this code:

func cancel() {
  complete(with: .finished)
}

As with a subscriber, you’ll need to implement both methods that accept values and a completion event. Start by adding this method to accept values:

func receive(_ input: Output) {
  guard subscriber != nil else { return }
  // 17
  buffer.append(input)
  if buffer.count > capacity {
    // 18
    buffer.removeFirst()
  }
  // 19
  emitAsNeeded()
}

After ensuring there is a subscriber, this method will:

17. Add the value to the outstanding buffer. You could optimize this for most common cases, such as unlimited demands, but this will do the job perfectly for now.
18. Make sure not to buffer more values than the requested capacity. You handle this on a rolling, first-in-first-out basis – as an already-full buffer receives each new value, the current first value is removed.
19. Deliver the results to the subscriber.

Wrapping up your subscription

Now, add the following method to accept completion events and your subscription class will be complete:

func receive(completion: Subscribers.Completion<Failure>) {
  guard let subscriber = subscriber else { return }
  self.subscriber = nil
  self.buffer.removeAll()
  subscriber.receive(completion: completion)
}

This method removes the subscriber, empties the buffer – because that’s just good memory management – and sends the completion downstream.

You’re done with the subscription! Isn’t this fun? Now, it’s time to code the publisher.

Coding your publisher

Publishers are usually value types (struct) in the Publishers namespace. Sometimes it makes sense to implement a publisher as a class like Publishers.Multicast, which multicast() returns, or Publishers.Share which share() returns. For this publisher, you’ll need a class, similarly to share(). This is the exception to the rule, though, as most often you’ll use a struct.

Start by adding this code to define your publisher class after your subscription:

extension Publishers {
  // 20
  final class ShareReplay<Upstream: Publisher>: Publisher {
    // 21
    typealias Output = Upstream.Output
    typealias Failure = Upstream.Failure
  }
}

20. You want multiple subscribers to be able to share a single instance of this operator, so you use a class instead of a struct. It’s also generic, with the final type of the upstream publisher as a parameter.
21. This new publisher doesn’t change the output or failure types of the upstream publisher – it simply uses the upstream’s types.

Adding the publisher’s required properties

Now, add the properties your publisher will need to the definition of ShareReplay:

// 22
private let lock = NSRecursiveLock()
// 23
private let upstream: Upstream
// 24
private let capacity: Int
// 25
private var replay = [Output]()
// 26
private var subscriptions = [ShareReplaySubscription<Output, Failure>]()
// 27
private var completion: Subscribers.Completion<Failure>? = nil

What this code does: 22. Because you’re going to be feeding multiple subscribers at the same time, you’ll need a lock to guarantee exclusive access to your mutable variables.

1. Keeps a reference to the upstream publisher. You’ll need it at various points in the subscription lifecycle.
2. You specify the maximum recording capacity of your replay buffer during initialization.
3. Naturally, you’ll also need storage for the values you record.
4. You feed multiple subscribers, so you’ll need to keep them around to notify them of events. Each subscriber gets its values from a dedicated ShareReplaySubscription — you’re going to code this in a short while.
5. The operator can replay values even after completion, so you need to remember whether the upstream publisher completed.

Phew! By the look of it, there’s some more code to write! In the end, you’ll see it’s not that much, but there is housekeeping to do, like using proper locking, so that your operator will run smoothly under all conditions.

Initializing and relaying values to your publisher

Firstly, add the necessary initializer to your ShareReplay publisher:

init(upstream: Upstream, capacity: Int) {
  self.upstream = upstream
  self.capacity = capacity
}

Nothing fancy here, just storing the upstream publisher and the capacity. Next, you’ll add a couple of methods to help split the code into smaller chunks.

Add the method that relays incoming values from upstream to subscribers:

private func relay(_ value: Output) {
  // 28
  lock.lock()
  defer { lock.unlock() }

  // 29
  guard completion == nil else { return }

  // 30
  replay.append(value)
  if replay.count > capacity {
    replay.removeFirst()
  }
  // 31
  subscriptions.forEach {
    $0.receive(value)
  }
}

This code does the following:

28. Since multiple subscribers share this publisher, you must protect access to mutable variables with locks. Using defer here is not strictly needed, but it’s good practice just in case you later modify the method, add an early return statement and forget to unlock your lock.
29. Only relays values if the upstream hasn’t completed yet.
30. Adds the value to the rolling buffer and only keeps the latest values of capacity. These are the ones to replay to new subscribers.
31. Relays the buffered values to each connected subscriber.

Letting your publisher know when it’s done

Secondly, add this method to handle completion events:

private func complete(_ completion: Subscribers.Completion<Failure>) {
  lock.lock()
  defer { lock.unlock() }
  // 32
  self.completion = completion
  // 33
  subscriptions.forEach {
    $0.receive(completion: completion)
  }
}

With this code, you’re:

32. Saving the completion event for future subscribers.
33. Relaying it to each connected subscriber.

You are now ready to start coding the receive method that every publisher must implement. This method will receive a subscriber. Its duty is to create a new subscription and then hand it over to the subscriber.

Add this code to begin defining this method:

func receive<S: Subscriber>(subscriber: S)
  where Failure == S.Failure,
        Output == S.Input {
  lock.lock()
  defer { lock.unlock() }
}

This standard prototype for receive(subscriber:) specifies that the subscriber, whatever it is, must have Input and Failure types that match the publisher’s Output and Failure types. Remember this from Chapter 2, “Publishers & Subscribers?”

Creating your subscription

Next, add this code to the method to create the subscription and hand it over to the subscriber:

// 34
let subscription = ShareReplaySubscription(
  subscriber: subscriber,
  replay: replay,
  capacity: capacity,
  completion: completion)

// 35
subscriptions.append(subscription)
// 36
subscriber.receive(subscription: subscription)

34. The new subscription references the subscriber and receives the current replay buffer, the capacity, and any outstanding completion event.
35. You keep the subscription around to pass future events to it.
36. You send the subscription to the subscriber, which may — either now or later — start requesting values.

Subscribing to the upstream publisher and handling its inputs

You are now ready to subscribe to the upstream publisher. You only need to do it once: When you receive your first subscriber.

Add this code to receive(subscriber:) – note that you are intentionally not including the closing } because there’s more code to add:

// 37
guard subscriptions.count == 1 else { return }

let sink = AnySubscriber(
  // 38
  receiveSubscription: { subscription in
    subscription.request(.unlimited)
  },
  // 39
  receiveValue: { [weak self] (value: Output) -> Subscribers.Demand in
    self?.relay(value)
    return .none
  },
  // 40
  receiveCompletion: { [weak self] in
    self?.complete($0)
  }
)

With this code you:

37. Subscribe only once to the upstream publisher.
38. Use the handy AnySubscriber class which takes closures, and immediately request .unlimited values upon subscription to let the publisher run to completion.
39. Relay values you receive to downstream subscribers.
40. Complete your publisher with the completion event you get from upstream.

Note: You could initially request .max(self.capacity) and receive just that, but remember that Combine is demand-driven! If you don’t request as many values as the publisher is capable of producing, you may never get a completion event!

To avoid retain cycles, you only keep a weak reference to self.

You’re nearly done! Now, all you need to do is subscribe AnySubscriber to the upstream publisher.

Finish off the definition of this method by adding this code:

upstream.subscribe(sink)

Once again, all errors in the playground should be clear now. Remember that you can double-check your work by comparing it with the finished version of the playground in projects/final.

Adding a convenience operator

Your publisher is complete! Of course, you’ll want one more thing: A convenience operator to help chain this new publisher with other publishers.

Add it as an extension to the Publishers namespace at the end of your playground:

extension Publisher {
  func shareReplay(capacity: Int = .max)
    -> Publishers.ShareReplay<Self> {
    return Publishers.ShareReplay(upstream: self,
                                  capacity: capacity)
  }
}

You now have a fully functional shareReplay(capacity:) operator. This was a lot of code, and now it’s time to try it out!

Testing your subscription

Add this code to the end of your playground to test your new operator:

// 41
var logger = TimeLogger(sinceOrigin: true)
// 42
let subject = PassthroughSubject<Int,Never>()
// 43
let publisher = subject.shareReplay(capacity: 2)
// 44
subject.send(0)

Here’s what this code does:

41. Use the handy TimeLogger object defined in this playground.
42. To simulate sending values at different times, you use a subject.
43. Share the subject and replay the last two values only.
44. Send an initial value through the subject. No subscriber has connected to the shared publisher, so you shouldn’t see any output.

Now, create your first subscription and send some more values:

let subscription1 = publisher.sink(
  receiveCompletion: {
    print("subscription1 completed: \($0)", to: &logger)
  },
  receiveValue: {
    print("subscription1 received \($0)", to: &logger)
  }
)

subject.send(1)
subject.send(2)
subject.send(3)

Next, create a second subscription and send a couple more values and then a completion event:

let subscription2 = publisher.sink(
  receiveCompletion: {
    print("subscription2 completed: \($0)", to: &logger)
  },
  receiveValue: {
    print("subscription2 received \($0)", to: &logger)
  }
)

subject.send(4)
subject.send(5)
subject.send(completion: .finished)

These two subscriptions display every event they receive, along with the time that’s elapsed since start.

Add one more subscription with a small delay to make sure it occurs after the publisher has completed:

var subscription3: Cancellable? = nil

DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
  print("Subscribing to shareReplay after upstream completed")
  subscription3 = publisher.sink(
    receiveCompletion: {
      print("subscription3 completed: \($0)", to: &logger)
    },
    receiveValue: {
      print("subscription3 received \($0)", to: &logger)
    }
  )
}

Remember that a subscription terminates when it’s deallocated, so you’ll want to use a variable to keep the deferred one around. The one-second delay demonstrates how the publisher replays data in the future. You’re ready to test! Run the playground to see the following results in the debug console:

+0.02967s: subscription1 received 1
+0.03092s: subscription1 received 2
+0.03189s: subscription1 received 3
+0.03309s: subscription2 received 2
+0.03317s: subscription2 received 3
+0.03371s: subscription1 received 4
+0.03401s: subscription2 received 4
+0.03515s: subscription1 received 5
+0.03548s: subscription2 received 5
+0.03716s: subscription1 completed: finished
+0.03746s: subscription2 completed: finished
Subscribing to shareReplay after upstream completed
+1.12007s: subscription3 received 4
+1.12015s: subscription3 received 5
+1.12057s: subscription3 completed: finished

Your new operator is working beautifully:

• The 0 value never appears in the logs, because it was emitted before the first subscriber subscribed to the shared publisher.
• Every value propagates to current and future subscribers.
• You created subscription2 after three values have passed through the subject, so it only sees the last two (values 2 and 3)
• You created subscription3 after the subject has completed, but the subscription still received the last two values that the subject emitted.
• The completion event propagates correctly, even if the subscriber comes after the shared publisher has completed.

Verifying your subscription

Fantastic! This works exactly as you wanted. Or does it? How can you verify that the publisher is being subscribed to only once? By using the print(_:) operator, of course! You can try it by inserting it before shareReplay.

Find this code:

let publisher = subject.shareReplay(capacity: 2)

And change it to:

let publisher = subject
  .print("shareReplay")
  .shareReplay(capacity: 2)

Run the playground again and it will yield this output:

shareReplay: receive subscription: (PassthroughSubject)
shareReplay: request unlimited
shareReplay: receive value: (1)
+0.03004s: subscription1 received 1
shareReplay: receive value: (2)
+0.03146s: subscription1 received 2
shareReplay: receive value: (3)
+0.03239s: subscription1 received 3
+0.03364s: subscription2 received 2
+0.03374s: subscription2 received 3
shareReplay: receive value: (4)
+0.03439s: subscription1 received 4
+0.03471s: subscription2 received 4
shareReplay: receive value: (5)
+0.03577s: subscription1 received 5
+0.03609s: subscription2 received 5
shareReplay: receive finished
+0.03759s: subscription1 received completion: finished
+0.03788s: subscription2 received completion: finished
Subscribing to shareReplay after upstream completed
+1.11936s: subscription3 received 4
+1.11945s: subscription3 received 5
+1.11985s: subscription3 received completion: finished

All the lines beginning with shareReplay are logs showing what happens with the original subject. Now you are sure that it’s performing the work only once and sharing the results with all current and future subscribers. Job very well done!

This chapter taught you several techniques to create your own publishers. It’s been long and complex, as there was quite some code to write. You’re nearly done now, but there’s one last topic you’ll want to learn about before moving on.

Handling backpressure

In fluid dynamics, backpressure is a resistance or force opposing the desired flow of fluid through pipes. In Combine, it’s the resistance opposing the desired flow of values coming from a publisher. But what is this resistance? Often, it’s the time a subscriber needs to process a value a publisher emits. Some examples are:

• Processing high-frequency data, like input from sensors.
• Performing large file transfers.
• Rendering complex UI upon data update.
• Waiting for user input.
• More generally, processing incoming data that the subscriber can’t keep up with at the rate it’s coming in.

The publisher-subscriber mechanism offered by Combine is flexible. It is a pull design, as opposed to a push one. It means that subscribers ask publishers to emit values and specify how many they want to receive. This request mechanism is adaptive: The demand updates every time the subscriber receives a new value. This allows subscribers to deal with backpressure by “closing the tap” when they don’t want to receive more data, and “opening it” later when they are ready for more.

Note: Remember, you can only adjust demand in an additive way. You can increase demand each time the subscriber receives a new value, by returning a new .max(N) or .unlimited. Or you can return .none, which indicates that the demand should not increase. However, the subscriber is then “on the hook” to receive values at least up to the new max demand. For example, if the previous max demand was to receive three values and the subscriber has only received one, returning .none in the subscriber’s receive(_:) will not “close the tap.” The subscriber will still receive at most two values when the publisher is ready to emit them.

What happens when more values are available is totally up to your design. You can:

• Control the flow by managing demand to prevent the publisher from sending more values than you can handle.
• Buffer values until you can handle them — with the risk of exhausting available memory.
• Drop values you can’t handle right away.
• Some combination of the above, according to your requirements.

Going through all possible combinations and implementations could take several chapters. In addition to the above, dealing with backpressure can take the form of:

• A publisher with a custom Subscription dealing with congestion.
• A subscriber delivering values at the end of a chain of publishers.

In this introduction to backpressure management, you’ll focus on implementing the latter. You’re going to create a pausable variant of the sink function, which you already know well.

Using a pausable sink to handle backpressure

To get started, switch to the PausableSink page of the playground.

As a first step, create a protocol that lets you resume from a pause:

protocol Pausable {
  var paused: Bool { get }
  func resume()
}

You don’t need a pause() method here, since you’ll determine whether or not to pause when you receive each value. Of course, a more elaborate pausable subscriber could have a pause() method you can call at any time! For now, you’ll keep the code as simple and straightforward as possible.

Next, add this code to start defining the pausable Subscriber:

// 1
final class PausableSubscriber<Input, Failure: Error>:
  Subscriber, Pausable, Cancellable {
  // 2
  let combineIdentifier = CombineIdentifier()
}

1. Your pausable subscriber is both Pausable and Cancellable. This is the object your pausableSink function will return. This is also why you implement it as a class and not as a struct: You don’t want an object to be copied, and you need mutability at certain points in its lifetime.
2. A subscriber must provide a unique identifier for Combine to manage and optimize its publisher streams.

Now add these additional properties:

// 3
let receiveValue: (Input) -> Bool
// 4
let receiveCompletion: (Subscribers.Completion<Failure>) -> Void

// 5
private var subscription: Subscription? = nil
// 6
var paused = false

3. The receiveValue closure returns a Bool: true indicates that it may receive more values and false indicates the subscription should pause.
4. The completion closure will be called upon receiving a completion event from the publisher.
5. Keep the subscription around so that it can request more values after a pause. You need to set this property to nil when you don’t need it anymore to avoid a retain cycle.
6. You expose the paused property as per the Pausable protocol.

Next, add the following code to PausableSubscriber to implement the initializer and to conform to the Cancellable protocol:

// 7
init(receiveValue: @escaping (Input) -> Bool,
     receiveCompletion: @escaping (Subscribers.Completion<Failure>) -> Void) {
  self.receiveValue = receiveValue
  self.receiveCompletion = receiveCompletion
}

// 8
func cancel() {
  subscription?.cancel()
  subscription = nil
}

7. The initializer accepts two closures, which the subscriber will call upon receiving a new value from the publisher and upon completion. The closures are like the ones you use with the sink function, with one exception: The receiveValue closure returns a Boolean to indicate whether the receiver is ready to take more values or whether you need to put the subscriptions on hold.
8. When canceling the subscription, don’t forget to set it to nil afterwards to avoid retain cycles.

Now add this code to satisfy Subscriber’s requirements:

func receive(subscription: Subscription) {
  // 9
  self.subscription = subscription
  // 10
  subscription.request(.max(1))
}

func receive(_ input: Input) -> Subscribers.Demand {
  // 11
  paused = receiveValue(input) == false
  // 12
  return paused ? .none : .max(1)
}

func receive(completion: Subscribers.Completion<Failure>) {
  // 13
  receiveCompletion(completion)
  subscription = nil
}

9. Upon receiving the subscription created by the publisher, store it for later so that you’ll be able to resume from a pause.
10. Immediately request one value. Your subscriber is pausable and you can’t predict when a pause will be needed. The strategy here is to request values one by one.
11. When receiving a new value, call receiveValue and update the paused status accordingly.
12. If the subscriber is paused, returning .none indicates that you don’t want more values right now — remember, you initially requested only one. Otherwise, request one more value to keep the cycle going.
13. Upon receiving a completion event, forward it to receiveCompletion then set the subscription to nil since you don’t need it anymore.

Finally, implement the rest of Pausable:

func resume() {
  guard paused else { return }

  paused = false
  // 14
  subscription?.request(.max(1))
}

14. If the publisher is “paused”, request one value to start the cycle again.

Just as you did with previous publishers, you can now expose your new pausable sink in the Publishers namespace.

Add this code at the end of your playground:

extension Publisher {
  // 15
  func pausableSink(
    receiveCompletion: @escaping ((Subscribers.Completion<Failure>) -> Void),
    receiveValue: @escaping ((Output) -> Bool))
    -> Pausable & Cancellable {
    // 16
    let pausable = PausableSubscriber(
      receiveValue: receiveValue,
      receiveCompletion: receiveCompletion)
    self.subscribe(pausable)
    // 17
    return pausable
  }
}

15. Your pausableSink operator is very close to the sink operator. The only difference is the return type for the receiveValue closure: Bool.
16. Instantiate a new PausableSubscriber and subscribe it to self, the publisher.
17. The subscriber is the object you’ll use to resume and cancel the subscription.

Testing your new sink

You can now try your new sink! To make things simple, simulate cases where the publisher should stop sending values. Add this code:

let subscription = [1, 2, 3, 4, 5, 6]
  .publisher
  .pausableSink(receiveCompletion: { completion in
    print("Pausable subscription completed: \(completion)")
  }) { value -> Bool in
    print("Receive value: \(value)")
    if value % 2 == 1 {
      print("Pausing")
      return false
    }
    return true
}

An array’s publisher usually emits all its values sequentially, one right after the other. Using your pausable sink, this publisher will pause when values 1, 3 and 5 are received.

Run the playground and you’ll see:

Receive value: 1
Pausing

To resume the publisher, you need to call resume() asynchronously. This is easy to do with a timer. Add this code to set up a timer:

let timer = Timer.publish(every: 1, on: .main, in: .common)
  .autoconnect()
  .sink { _ in
    guard subscription.paused else { return }
    print("Subscription is paused, resuming")
    subscription.resume()
  }

Run the playground again and you’ll see the pause/resume mechanism in action:

Receive value: 1
Pausing
Subscription is paused, resuming
Receive value: 2
Receive value: 3
Pausing
Subscription is paused, resuming
Receive value: 4
Receive value: 5
Pausing
Subscription is paused, resuming
Receive value: 6
Pausable subscription completed: finished

Congratulations! You now have a functional pausable sink and you’ve gotten a glimpse into handling backpressure in your code!

Note: What if your publisher can’t hold values and wait for the subscriber to request them? In this situation, you’d want to buffer values using the buffer(size:prefetch:whenFull:) operator. This operator can buffer values up to the capacity you indicate in the size parameter and deliver them when the subscriber is ready to receive them. The other parameters determine how the buffer fills up – either at once when subscribing, keeping the buffer full, or upon request from its subscriber – and what happens when the buffer is full – i.e., drop the last value(s) it received, drop the oldest one(s) or terminate with an error.

Key points

Wow, this was a long and complex chapter! You learned a lot about publishers:

• A publisher can be a simple method that leverages other publishers for convenience.
• Writing a custom publisher usually involves creating an accompanying Subscription.
• The Subscription is the real link between a Subscriber and a Publisher.
• In most cases, the Subscription is the one that does all the work.
• A Subscriber can control the delivery of values by adjusting its Demand.
• The Subscription is responsible for respecting the subscriber’s Demand. Combine does not enforce it, but you definitely should respect it as a good citizen of the Combine ecosystem.

Where to go from here?

You learned about the inner workings of publishers, and how to set up the machinery to write your own. Of course, any code you write — and publishers in particular! — should be thoroughly tested. Move on to the next chapter to learn all about testing Combine code!