11. Time-Based Operators

Written by Florent Pillet

Timing is everything. The core idea behind reactive programming is to model asynchronous data flow over time. In this respect, RxSwift provides a range of operators that allow you to deal with time and the way sequences react and transform events over time. As you’ll see throughout this chapter, managing the time dimension of your sequences is easy and straightforward.

To learn about time-based operators, you’ll practice with an animated playground that demonstrates visually how data flows over time. This chapter comes with an empty RxSwiftPlayground, divided in several pages. You’ll use each page to exercise one or more related operators. The playground also includes a number of ready-made classes that’ll come in handy to build the examples.

Getting started

For this chapter, you will be using an Xcode Playground that‘s been set up with the basic building blocks you need to go through the chapter tasks.

To get started, open the macOS Terminal application (found in your Mac’s Applications/Utilities folder), navigate to the current chapter’s starter project folder, then run the bootstrap script like so:

$ ./bootstrap.sh

You can keep the Debug Area visible, but what is most important is that you show the Live View pane. This will display a live view of the sequences you build in code. This is where the real action will happen!

To display the Live View, click the second-to-last button at top-right of the Xcode window’s editor (right below the title bar), as shown below:

The icon of this button changes depending on whether the live view is already visible.

Also make sure that anything you type automatically executes in the Assistant Editor’s preview area. Long-click the play/stop button at the bottom of the editor (if it is currently set to run, it will be a square) and make sure Automatically Run is selected, as in the screenshot below:

In the left Navigator pane, pick the first page named replay. You can then close the Navigator pane using its visibility control, which is the leftmost button at the top-right of the Xcode window.

Your layout should now look like this:

You’re now all set! It’s time to learn about the first group of time-based operators: buffering operators.

Note: This playground uses advanced features of Xcode playgrounds. Xcode does not fully support importing linked frameworks from within files in the common Sources subfolder. Therefore, each playground page has to include a small bit of code (the part of TimelineView that depends on RxSwift) to function properly. Just ignore this code and leave it at bottom of the page.

Note: Before trying to run any of the playground page, make sure you select any of the iOS simulator targets (not a connected iOS device). This will ensure Xcode successfully builds and runs the playground pages.

Buffering operators

The first group of time-based operators deal with buffering. They will either replay past elements to new subscribers, or buffer them and deliver them in bursts. They allow you to control how and when past and new elements get delivered.

Replaying past elements

When a sequence emits items, you’ll often need to make sure that a future subscriber receives some or all of the past items. This is the purpose of the replay(_:) and replayAll() operators. To learn how to use them, you’ll start coding in the replay page of the playground. To visualize what replay(_:) does, you’ll display elements on a timeline. The playground contains custom classes to make it easy to display animated timelines.

Start by adding some definitions:

let elementsPerSecond = 1
let maxElements = 58
let replayedElements = 1
let replayDelay: TimeInterval = 3

You’ll create an observable that emits elements at a frequency of elementsPerSecond. You’ll also cap the total number of elements emitted, and control how many elements are “played back” to new subscribers.

To build this emitting observable, use the Observable.create method and some dispatch magic:

let sourceObservable = Observable<Int>.create { observer in
  var value = 1
  let timer = DispatchSource.timer(interval: 1.0 / Double(elementsPerSecond), queue: .main) {

The DispatchSource.timer function is an extension to DispatchSource defined in the playground’s Sources folder. It simplifies the creation of repeating timers. Add the code to emit elements:

    if value <= maxElements {
      observer.onNext(value)
      value += 1
    }
  }
  return Disposables.create {
    timer.suspend()
  }
}

Note that for the purpose of this example, you don’t care about completing the observable. It simply emits as many elements as instructed and never completes.

Now add the replay functionality to the observable, by appending it to the end of the sourceObservable chain:

.replay(replayedElements)

This operator creates a new sequence which records the last replayedElements emitted by the source observable. Every time a new observer subscribes, it immediately receives the buffered elements, if any, and keeps receiving any new element like a normal observer does.

To visualize the actual effect of replay(_:), create a couple of TimelineView views. This class is defined at bottom of the playground page and relies on the TimelineViewBase class in the Sources group of the playground. It provides a live visualization of events emitted by an observable. Append, below the code you just wrote:

let sourceTimeline = TimelineView<Int>.make()
let replayedTimeline = TimelineView<Int>.make()

You’re going to use a UIStackView for convenience. It’ll display the source (live) observable as viewed by an immediate subscriber, as well as another representation as viewed by a subscriber coming later. Create the stack view:

let stack = UIStackView.makeVertical([
  UILabel.makeTitle("replay"),
  UILabel.make("Emit \(elementsPerSecond) per second:"),
  sourceTimeline,
  UILabel.make("Replay \(replayedElements) after \(replayDelay) sec:"),
  replayedTimeline])

This looks complicated, but it’s actually fairly straightforward. It simply creates a few vertically stacked views. The UIStackView.makeVertical(_:) and UILabel.make(_:) methods are convenience extensions local to this playground.

Next, prepare an immediate subscriber and display what it receives in the top timeline:

_ = sourceObservable.subscribe(sourceTimeline)

The TimelineView class implements RxSwift‘s ObserverType protocol. Therefore, you can subscribe it to an observable sequence and it will receive the sequence’s events. Every time a new event occurs (element emitted, sequence completed or errored out), TimelineView displays it on the timeline. Emitted elements are shown in green, completion in black and error in red.

Note: Did you notice that the code is ignoring the Disposable returned by the subscription? Good! This example code is not keeping them on purpose, as the playground page drops everything when refreshing. In your applications, remember to always keep long-running subscriptions in a DisposeBag!

Next, you want to subscribe again to the source observable, but with a slight delay:

DispatchQueue.main.asyncAfter(deadline: .now() + replayDelay) {
  _ = sourceObservable.subscribe(replayedTimeline)
}

This displays elements received by the second subscription in another timeline view. You’ll see the timeline view shortly, I promise!

Since replay(_:) creates a connectable observable, you need to connect it to its underlying source to start receiving items. If you forget this, subscribers will never receive anything.

Note: Connectable observables are a special class of observables. Regardless of their number of subscribers, they won‘t start emitting items until you call their connect() method. While this is beyond the scope of this chapter, remember that a few operators return ConnectableObservable<E>, not Observable<E>. These operators are:

replay(_:)

replayAll()

multicast(_:)

publish()

Replay operators are covered in this chapter. The last two operators are advanced, and only touched on briefly in this book. They allow sharing a single subscription to an observable, regardless of the number of observers.

So add this code to connect:

_ = sourceObservable.connect()

Finally, set up the host view in which the stack view will display. The playground has a utility function to keep your code simple:

let hostView = setupHostView()
hostView.addSubview(stack)
hostView

Once you save these source changes, Xcode will recompile the playground code and… look at the Live View pane! Finally!

You’ll see two timelines. The top timeline reflects an immediate subscription to sourceObservable. The bottom timeline is the one where the subscription occurs after a delay. The source observable emits numbers for convenience. This way you can see the progress of emitted elements.

You may need to wait a little bit after making changes to the playground to have the timeline view show up, especially on slower computers. Such is life with Xcode.

Note: As exciting it is to see a live observable diagram, it might confuse at first. Static timelines usually have their elements aligned to the left, but if you think about it, they also have the most recent ones on the right side just as the animated diagrams you observe right now.

In the settings you used, replayedElements is equal to 1. It configures the replay(_:) operator to only buffer the last element from the source observable. The animated timeline shows that the second subscriber receives elements 3 and 4 in about the same time frame. Depending on your system load, dispatch may incur a slight delay, producing a little variation on the screenshot above.

By the time it subscribes, it gets both the latest buffered element (3) and the one that happens to be emitted just right when subscription occurs. The timeline view shows them stacked up since the time they arrive is about the same, although not exactly the same.

Note: You can now play with the replayDelay and replayedElements constants. Observe the effect of tweaking the number of replayed (buffered) elements. You can also tweak the total number of elements emitted by the source observable using maxElements. Set it to a very large value for continuous emission.

Unlimited replay

The second replay operator you can use is replayAll(). This one should be used with caution: only use it in scenarios where you know the total number of buffered elements will stay reasonable. For example, it’s appropriate to use replayAll() in the context of HTTP requests. You know the approximate memory impact of retaining the data returned by a query. On the other hand, using replayAll() on a sequence that may not terminate and may produce a lot of data will quickly clog your memory. This could grow to the point where the OS jettisons your application!

To experiment with replayAll(), replace:

.replay(replayedElements)

with:

.replayAll()

Watch the effect on the timeline. You will see all buffered elements emitted instantly upon the second subscription.

Controlled buffering

Now that you touched on replayable sequences, you can look at a more advanced topic: controlled buffering. You’ll first look at the buffer(timeSpan:count:scheduler:) operator. Switch to the second page in the playground called buffer. As in the previous example, you’ll begin with some constants:

let bufferTimeSpan: RxTimeInterval = .seconds(4)
let bufferMaxCount = 2

These constants define the behavior for the buffer operator you‘ll soon add to the code. For this example, you’ll manually feed a subject with values. Add:

let sourceObservable = PublishSubject<String>()

You will push short strings (a single emoji) to this observable. Create the timeline visualizations and the stack to contain them just like before:

let sourceTimeline = TimelineView<String>.make()
let bufferedTimeline = TimelineView<Int>.make()

let stack = UIStackView.makeVertical([
  UILabel.makeTitle("buffer"),
  UILabel.make("Emitted elements:"),
  sourceTimeline,
  UILabel.make("Buffered elements (at most \(bufferMaxCount) every \(bufferTimeSpan) seconds):"),
  bufferedTimeline])

Subscribe to fill the top timeline with events, like you did in the replay playground page:

_ = sourceObservable.subscribe(sourceTimeline)

The buffered timeline will display the number of elements contained in each buffered array. Add the following code:

sourceObservable
  .buffer(timeSpan: bufferTimeSpan, count: bufferMaxCount, scheduler: MainScheduler.instance)
  .map(\.count)
  .subscribe(bufferedTimeline)

What’s happening here? Breaking it down:

• You want to receive arrays of elements from the source observable.
• Each array can hold at most bufferMaxCount elements.
• If that many elements are received before bufferTimeSpan expires, the operator will emit buffered elements and reset its timer.
• In a delay of bufferTimeSpan after the last emitted group, buffer will emit an array. If no element has been received during this timeframe, the array will be empty.

To activate your timeline views, set up the host view:

let hostView = setupHostView()
hostView.addSubview(stack)
hostView

Even though there is no activity on the source observable, you can witness empty buffers on the buffered timeline. The buffer(_:scheduler:) operators emits empty arrays at regular intervals if nothing has been received from its source observable. The 0s mean that zero elements have been emitted from the source sequence.

You can start feeding the raw observable with data and observe the impact on the buffered observable. First, try pushing three elements over five seconds. Append:

DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
  sourceObservable.onNext("🐱")
  sourceObservable.onNext("🐱")
  sourceObservable.onNext("🐱")
}

Can you guess what the effect will be? Look how the timeline moves:

Each box shows the number of elements in each emitted array:

• At first, the buffered timeline emits an empty array. There’s no element in the source observable yet.
• Then, you push three elements on the source observable.
• The buffered timeline immediately gets an array of two elements because it’s the maximum count you specified (due to the bufferMaxCount constant).
• Four seconds elapse, and an array with just one element is emitted. This is the last of the three elements that have been pushed to the source observable.

As you can see, the buffer immediately emits an array of elements when it reaches full capacity, then waits for the specified delay, or until it’s full again, before it emits a new array.

You can play a bit more with different buffering scenarios.

Remove the DispatchQueue that emits elements, and add this instead:

let elementsPerSecond = 0.7
let timer = DispatchSource.timer(interval: 1.0 / Double(elementsPerSecond), queue: .main) {
  sourceObservable.onNext("🐱")
}

The timeline is very different! As before, you can tweak the constants (buffering time, buffering limit, elements per second) to see how grouping works.

Windows of buffered observables

A last buffering technique very close to buffer(timeSpan:count:scheduler:) is window(timeSpan:count:scheduler:). It has roughly the same signature and does nearly the same thing. The only difference is that it emits an Observable of the buffered items, instead of emitting an array.

You’re going to build a slightly more elaborate timeline view. Since windowed sequences emit multiple observables, it will be beneficial to visualize them separately. Get started in the window playground page:

let elementsPerSecond = 3
let windowTimeSpan: RxTimeInterval = .seconds(4)
let windowMaxCount = 10
let sourceObservable = PublishSubject<String>()

You‘re going to look at how timed output is grouped in windowed observables by pushing strings to a subject. As usual, first add the stack view code:

let sourceTimeline = TimelineView<String>.make()

let stack = UIStackView.makeVertical([
  UILabel.makeTitle("window"),
  UILabel.make("Emitted elements (\(elementsPerSecond) per sec.):"),
  sourceTimeline,
  UILabel.make("Windowed observables (at most \(windowMaxCount) every \(windowTimeSpan) sec):")])

This time, add a timer to push elements to the source observable:

let timer = DispatchSource.timer(interval: 1.0 / Double(elementsPerSecond), queue: .main) {
  sourceObservable.onNext("🐱")
}

Then fill up the source timeline:

_ = sourceObservable.subscribe(sourceTimeline)

You’re now at a point where you want to see each emitted observable separately. To this end, you’ll insert a new timeline every time window(timeSpan:count:scheduler:) emits a new observable. Previous observables will move downwards. Append:

_ = sourceObservable
  .window(timeSpan: windowTimeSpan, count: windowMaxCount, scheduler: MainScheduler.instance)

This is your windowed observable. How can you handle emitted observables? Using your trusted flatMap(_:) operator of course! Chain this under the window operator:

.flatMap { windowedObservable -> Observable<(TimelineView<Int>, String?)> in
  let timeline = TimelineView<Int>.make()
  stack.insert(timeline, at: 4)
  stack.keep(atMost: 8)
  return windowedObservable
    .map { value in (timeline, value) }
    .concat(Observable.just((timeline, nil)))
}

Obviously this is the tricky part. Try to figure out the code yourself first, and then fall back on the following:

• Every time flatMap(_:) gets a new observable, you insert a new timeline view.
• You then map the observable of items to an observable of tuple. The goal is to transport both the value and the timeline in which to display it.
• Once this inner observable completes, you concat(_:) a single tuple so you can mark the timeline as complete.
• You flatMap(_:) the sequence of resulting observables of tuple to a single sequence of tuples.
• You subscribe to the resulting observable and fill up timelines as you receive tuples.

Note: In trying to keep the code short, you’re doing something that is generally not advisable in Rx code: you’re adding side effects to an operator that’s supposed to just be transforming data. The right solution would be to perform side effects using a do(onNext:) operator. This is left as an exercise in this chapter’s challenges!

Finally, you need to subscribe and display elements in each timeline. Since you mapped the elements to the actual timeline they belong to, this becomes easy. Chain this code to the previous:

.subscribe(onNext: { tuple in
  let (timeline, value) = tuple
  if let value = value {
    timeline.add(.next(value))
  } else {
    timeline.add(.completed(true))
  }
})

The value in the tuple is a String?: the convention here is that if it is nil, it means the sequence completed. The code pushes either a next or a completed event to the timeline.

Finally, instantiate the host view as usual:

let hostView = setupHostView()
hostView.addSubview(stack)
hostView

Let the playground run. Things quickly get interesting, as window(timeSpan:count:scheduler:) emits new sequences:

Starting from the second timeline, all the timelines you see are “most recent first”. This screenshot was taken with a setting of five elements maximum per windowed observable, and a four second window. This means that a new observable is produced at least every four seconds. It will emit at most five elements before completing.

If the source observable emits more than four elements during the window time, a new observable is produced, and the cycle starts again.

Time-shifting operators

Every now and again, you need to travel in time. While RxSwift can’t help with fixing your past relationship mistakes, it has the ability to freeze time for a little while to let you wait until self-cloning is available.

Next, you’ll look into two time related operators. Open the delay playground page to get started.

Delayed subscriptions

You’ll start with delaySubscription(_:scheduler:). Since you are now used to creating animated timelines, this page comes with most of the setup code ready. Find the comment Setup the delayed subscription in the source and insert this code below it:

_ = sourceObservable
  .delaySubscription(delay, scheduler: MainScheduler.instance)
  .subscribe(delayedTimeline)

The idea behind the delaySubscription(_:scheduler:) is, as the name implies, to delay the time a subscriber starts receiving elements from its subscription. Run the example if it’s not already running. In the right timeline view, you can observe that the second timeline starts picking up elements after the delay specified by delayInSeconds.

Note: In Rx, some observables are called “cold” while others are “hot”. Cold observables start emitting elements when you subscribe to them. Hot observables are more like permanent sources you happen to look at at some point (think of Notifications). When delaying a subscription, it won’t make a difference if the observable is cold. If it’s hot, you may skip elements, as in this example.

Hot and cold observables are a tricky topic that can take some time getting your head around. Remember that cold observables produce events only when subscribed to, but hot observables produce events independent of being subscribed to.

Delayed elements

The other kind of delay in RxSwift lets you time-shift the whole sequence. Instead of subscribing late, the operator subscribes immediately to the source observable, but delays every emitted element by the specified amount of time. The net result is a concrete time-shift.

To try this out, stay in the delay playground page you just used. Replace the delayed subscription (that you just added) with:

_ = sourceObservable
  .delay(delay, scheduler: MainScheduler.instance)
  .subscribe(delayedTimeline)

As you can see the code is similar. You just replaced delaySubscription(_:scheduler:) with delay(_:scheduler:). Look at the timelines. Can you spot the difference?

In the previous example, delaying the subscription (with the default settings) made you miss the first two elements from the source observable. When using the delay(_:scheduler:) operator, you time-shift the elements and won‘t miss any. Again, the subscription occurs immediately. You simply “see” the items with a delay.

Timer operators

A common need in any kind of application is a timer. iOS and macOS come with several timing solutions. Historically, Timer did the job, but had a confusing ownership model that made it tricky to get just right. More recently, the dispatch framework offered timers through the use of dispatch sources. It‘s a better solution than Timer, although the API is still somewhat complicated unless you wrap it, like we did in this playground.

RxSwift provides a simple and efficient solution for both one-shot and repeating timers. It integrates perfectly with sequences and offers both cancellation and composability with other sequences.

Intervals

This chapter used DispatchSource several times to create interval timers through a handy custom function. You could replace these instances with RxSwift’s Observable.interval(_:scheduler:) function. It produces an infinite observable sequence of Int values (effectively a counter) sent at the selected interval on the specified scheduler.

Go back to the replay playground page. Towards the beginning of the code, you created a source observable. You used DispatchSource.timer(interval:queue:) to create a timer and feed observers with values.

Delete this code, starting at let sourceObservable = Observable<Int>.create {... and up to (and including) replayAll(); and then insert instead:

let sourceObservable = Observable<Int>
  .interval(.milliseconds(Int(1000.0 / Double(elementsPerSecond))), scheduler: MainScheduler.instance)
  .replay(replayedElements)

And. That‘s. All.

Interval timers are incredibly easy to create with RxSwift. Not only that, but they are also easy to cancel: since Observable.interval(_:scheduler:) generates an observable sequence, you can simply dispose() the returned disposable to cancel the subscription and stop the timer. Very cool!

It is notable that the first value is emitted at the specified duration after a subscriber starts observing the sequence. Also, the timer won’t start before this point. The subscription is the trigger that kicks it off.

Note: As you can see in the timeline view, values emitted by Observable.interval(_:scheduler:) are signed integers starting from 0. Should you need different values, you can simply map(_:) them. In most real life cases, the value emitted by the timer is simply ignored. But it can make a convenient index.

One-shot or repeating timers

You may want a more powerful timer observable. You can use the Observable.timer(_:period:scheduler:) operator which is very much like Observable.interval(_:scheduler:) but adds the following features:

• You can specify a “due date” as the time that elapsed between the point of subscription and the first emitted value.
• The repeat period is optional. If you don’t specify one, the timer observable will emit once, then complete.

Can you see how handy this can be? Give it a go.

In the playground, open the delay page. Locate the place where you used the delay(_:scheduler:) operator. Replace the whole block of code with:

_ = Observable<Int>
  .timer(.seconds(3), scheduler: MainScheduler.instance)
  .flatMap { _ in
    sourceObservable.delay(delay, scheduler: MainScheduler.instance)
  }
  .subscribe(delayedTimeline)

A timer triggering another timer? This is Inception! There are several benefits to using this over Dispatch:

• The whole chain is more readable (more “Rx-y”).

• Since the subscription returns a disposable, you can cancel it at any point before the first or second timer triggers with a single observable.

• Using the flatMap(_:) operator, you can produce timer sequences without having to jump through hoops with Dispatch’s asynchronous closures.

Timeouts

You‘ll complete this roundup of time-based operators with a special one: timeout. Its primary purpose is to semantically distinguish an actual timer from a timeout (error) condition. Therefore, when a timeout operator fires, it emits an RxError.TimeoutError error event; if not caught, it terminates the sequence.

Open the timeout playground page. Create a simple button:

let button = UIButton(type: .system)
button.setTitle("Press me now!", for: .normal)
button.sizeToFit()

You’re going to use an extension from RxCocoa that turns button taps into an observable sequence. You’ll learn more about RxCocoa in the following chapters. For now, the goal is to:

• Capture button taps.
• If the button is pressed within five seconds, print something and terminate the sequence.
• If the button is not pressed, print the error condition.

Prepare the timeline view and stack it up with the button:

let tapsTimeline = TimelineView<String>.make()

let stack = UIStackView.makeVertical([
  button,
  UILabel.make("Taps on button above"),
  tapsTimeline])

Setup the observable and connect it to the timeline view:

let _ = button
  .rx.tap
  .map { _ in "•" }
  .timeout(5, scheduler: MainScheduler.instance)
  .subscribe(tapsTimeline)

And as usual, add the stack to the host view to kick off the animation:

let hostView = setupHostView()
hostView.addSubview(stack)
hostView

If you click the button within five seconds (and within five seconds of subsequent presses), you‘ll see your taps on the timeline. Stop clicking, and five seconds after that, as the timeout fires, the timeline will stop with an Error.

An alternate version of timeout(_:scheduler:) takes an observable and, when the timeout fires, switches the subscription to this observable instead of emitting an error.

There are many uses for this form of timeout, one of which is to emit a value (instead of an error) and then complete normally.

To try this, change the timeout(_:scheduler:) call in the playground to:

.timeout(5, other: Observable.just("X"), scheduler: MainScheduler.instance)

Now, instead of the error indicator, you see the X element and a regular completion. Mission accomplished!

Challenge

Challenge: Circumscribe side effects

In the discussion of the window(_:scheduler:) operator, you created timelines on the fly inside the closure of a flatMap(_:) operator. While this was done to keep the code short, one of the guidelines of reactive programming is to “not leave the monad”. In other words, avoid side effects except for specific areas created to apply side effects. Here, the “side effect” is the creation of a new timeline in a spot where only a transformation should occur.

Your task is to find an alternate way to do this. Try and pick the one that seems the most elegant to you. When finished, compare it with the proposed solution!

There are several possible approaches to tackle this challenge. The most effective will be to split the work into multiple observables then join them later.

Make the windowed observable a separate one that you use to produce two separate sequences: one that prepares the timeline views (remember that side effects can be performed with the do(onNext:) operator), and one that takes both the produced timeline view and the source sequence element (hint: use a combination of zip and flatMap) to generate a contextual value (timeline view and sequence) every time window emits a new sequence.