11. Time-Based Operators

Written by Alex Sullivan & Florent Pillet

Timing is everything. The core idea behind reactive programming is to model asynchronous data flow over time.

In this respect, RxJava provides a range of operators that allow you to deal with time and the way that sequences react and transform 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 app that demonstrates visually how data flows over time. This chapter comes with a basic app with several buttons that lead to different pages. You’ll use each page to exercise one or more related operators. The app also includes a number of ready-made classes that’ll come in handy to build the examples.

Getting started

Open the starter project for this section, then build and run the app. You should see a white screen with five gray buttons:

Clicking any of these buttons will send you to another screen that, for now, just has some text. As you work through this chapter, you’ll flesh out each page to demonstrate a different set of time-based reactive operators.

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 app. To visualize what replay does, you’ll display elements on a marble diagram-like view. The app contains custom classes to make it easy to display animated timelines.

Open ReplayActivity.kt.

Note: For this chapter, you’ll forgo the usual ViewModel + Activity approach takes elsewhere in the app. Since this app is meant to demonstrate different time-oriented operators, you don’t need to worry about using a proper architecture.

Start by adding some definitions above the class ReplayActivity : AppCompatActivity() line:

val elementsPerSecond = 1
val replayedElements = 1
val replayDelayInMs = 3500L
val maxElements = 5

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. Build this emitting Observable in onCreate(), by using the timer function:

val sourceObservable = Observable.create<Int> { emitter ->
  var value = 1
  val disposable = timer(elementsPerSecond) {
    if (value <= maxElements) {
      emitter.onNext(value)
      value++
    }
  }
}

The timer function is a helper function defined in TimerUtils.kt. It helps to create simple repeating timers. Feel free to look at its implementation, but it may not make sense until later in the chapter, when you cover the interval method. Suffice to say it uses RxJava under the hood and returns a Disposable.

In the lambda passed to the timer function, you’re using the emitter object to emit the next value and then incrementing value. At the end of the day, this Observable should now emit increasing values at the frequency you defined earlier.

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 end of the sourceObservable declaration:

.replay(replayedElements)

This operator creates a new sequence that records the last replayedElements number of elements emitted by the source Observable. Every time a new observer subscribes, it immediately receives these elements (if any) and then keeps receiving any new element like a normal subscription does.

To visualize the actual effects of replay, you’re going to use a custom UI widget created for this chapter called MarbleView. The MarbleView class shows elements as they’re emitted on a timeline, similar to the marble diagrams you’ve seen in previous chapters.

For this page, there are two MarbleViews already included in the activity_replay layout file. You’ll use these two views to visualize the replay operators.

Add the following at the bottom of the onCreate method:

sourceObservable.subscribe(replay_1)

replay_1 is the name of the first MarbleView for this screen.

The MarbleView class implements the Observer RxJava interface. 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), MarbleView displays it on the timeline.

Next, you want to subscribe again to the source Observable, but with a slight delay. Add the following again at the bottom of the onCreate method:

dispatchAfter(replayDelayInMs) {
  sourceObservable.subscribe(replay_2)
}

dispatchAfter is another special function to make it easier to perform one-off actions.

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

Now, 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 couple of operators return ConnectableObservable, not Observable. Specifically, the replay and publish operators.

Replay operators are covered in this chapter. The publish operator is advanced, and only touched on briefly in this book. It allows sharing a single subscription to an Observable, regardless of the number of observers.

So add this code to connect, at the end of onCreate():

sourceObservable.connect()

Now, build and run the app and navigate to the replay page.

You’ll see two timelines. The top marble view reflects an observer named connect() that subscribes before you.

The bottom marble view is the one where subscription occurs after a delay. The source Observable emits numbers for convenience.

This way you can see the progress of emitted elements.

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 twice 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 marble view shows that the second subscriber receives element 3 and then quickly receives element 4 afterwards. That’s because the 3 element was the one being replayed. Then the 4 element was emitted normally.

Try raising the replayedElements constant to two instead of one. You’ll see a much more noticeable impact on the MarbleView:

Since two elements were buffered, they were all emitted at the same time. The MarbleView class will group items emitted at (or around) the same time in a column to make them more visible.

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

In addition to the replay operator that takes in a maximum number of elements, there’s an overloaded version of replay that takes no arguments. If used with no arguments, the replay operator will ensure that every item in your Observable is replayed. 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 replay with no arguments 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 it 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 you see an OutOfMemoryException!

To experiment with this new behavior, replace:

.replay(replayedElements)

With:

.replay()

Watch the effect on the marble view. 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 operator. Switch to the second page in the app called BUFFER. As in the previous example, you’ll begin with some constants. Add the following to the top of the BufferActivity.kt file:

private val bufferMaxCount = 2
private val bufferTimeSpan = 4L

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. At the bottom of the onCreate method add:

val sourceObservable = PublishSubject.create<String>()

You will push short strings (a single emoji) to this Observable. You’ll again use two predefined MarbleView widgets contained in the activity_buffer.xml layout file.

Subscribe to fill the top marble view with events, like you did in the REPLAY page:

sourceObservable
  .subscribe(buffer_1)

The buffered marble view will display the number of elements contained in each buffered array:

sourceObservable
  .buffer(bufferTimeSpan, TimeUnit.SECONDS, bufferMaxCount)
  .map { it.size }
  .subscribe(buffer_2)

What’s happening here? Breaking it down:

• You want to receive lists of elements from the source Observable.
• Each list 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 a list. If no element has been received during this time frame, the list will be empty.

Try building and running the app and navigating to the BUFFER page now.

Even though there is no activity on the source Observable, you can witness empty buffers on the buffered marble view. The buffer operator emits empty lists at regular intervals if nothing has been received from its source Observable. The 0s mean that zero elements have been emitted from the source Observable.

You can start feeding the raw Observable with data and observe the impact on the buffered Observable. First, try pushing three elements after five seconds. Append this to the bottom of the onCreate() method:

dispatchAfter(5000) {
  sourceObservable.onNext("🐱")
  sourceObservable.onNext("🐱")
  sourceObservable.onNext("🐱")
}

Can you guess what the effect will be? Build and run, and look how the marble view moves:

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

• At first, the buffered marble view emits an empty array — there’s no element in the source Observable yet and the bufferTimeSpan amount of time has passed.
• Then you push three elements on the source Observable.
• The buffered marble view immediately gets an array of two elements because it’s the maximum count you specified (due to the bufferMaxCount constant).
• Four seconds elapse, and a list 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 dispatchAfter that emits elements, and add this instead:

val elementsPerSecond = 1

timer(elementsPerSecond) {
  sourceObservable.onNext("🐱")
}.addTo(disposables)

The marble view 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 is window. It has roughly the same signature and nearly does 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 MarbleView. Since windowed sequences emit multiple Observables, it will be beneficial to visualize them separately. Get started in the WindowActivity, which is the root of the WINDOW page by adding several constants to just after the class declaration of WindowActivity:

private val elementsPerSecond = 3
private val windowTimeSpan = 4L
private val windowMaxCount = 10L

You’re going to look at how timed output is grouped in windowed Observables by pushing strings to a subject. Start off by adding another PublishSubject<String> to the bottom of the onCreate() method:

val sourceObservable = PublishSubject.create<String>()

Now, add a timer to push new strings into the sourceObservable:

timer(elementsPerSecond) {
  sourceObservable.onNext("🐱")
}.addTo(disposables)

Then fill up the source marble view:

sourceObservable.subscribe(windowSource)

You’re now at a point where you want to see each emitted Observable separately. To this end, you’ll insert a new MarbleView every time window emits a new Observable.

Previous Observables will move downwards. Just before the end of onCreate(), append the following:

sourceObservable.window(windowTimeSpan, TimeUnit.SECONDS, AndroidSchedulers.mainThread(), windowMaxCount)

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 ->
  val marbleView = MarbleView(this)
  marble_views.addView(marbleView)
  windowedObservable
    .map { value -> value to marbleView}
    .concatWith(Observable.just("" to marbleView))
}

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 MarbleView into the already existing marble_views layout.
• You then map the Observable of items to an Observable of pairs. The goal is to transport both the value and the marble view in which to display it.
• Once this inner Observable completes, you concatWith a single pair with an empty first value, so you can mark the timeline as complete.
• You flatMap the sequence of resulting observables of pairs 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 doOnNext operator. This is left as an exercise in this chapter’s challenges!

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

.subscribe { (value, marbleView) ->
  if (value.isEmpty()) {
    marbleView.onComplete()
  } else {
    marbleView.onNext(value)
  }
}
.addTo(disposables)

The value in the tuple is a String: The convention here is that if it is empty, then it means the sequence completed. The code pushes either a next or a completed event to the marble view.

Build and run the app, and navigate to the window page. Things very quickly get interesting as new observables are emitted:

The C value at the bottom of some of the MarbleViews represent that Observable completing.

Starting from the second timeline, all the timelines you see are “most recent first.” This screenshot was taken with a setting of ten 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, ten elements before completing.

If the source Observable emits more than nine 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 RxJava 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. Navigate to DelayActivity.kt to get started.

Delayed subscriptions

Start off by adding the constants to the top of the class:

private val elementsPerSecond = 1
private val delayInSeconds = 3L

Next up, add a new PublishSubject at the bottom of the onCreate() method:

val sourceObservable = PublishSubject.create<Int>()

Now, add the code to add items to the sourceObservable below the previous declaration:

var current = 1
timer(elementsPerSecond) {
  sourceObservable.onNext(current)
  current++
}

And subscribe to sourceObservable with the source MarbleView:

sourceObservable.subscribe(source)

You’re going to start off the delay section by using the delaySubscription operator. Append the following:

sourceObservable
  .delaySubscription(delayInSeconds, TimeUnit.SECONDS, AndroidSchedulers.mainThread())
  .subscribe(delayed)

The idea behind the delaySubscription operator is, as the name implies, to delay the time a subscriber starts receiving elements from its subscription. Run the app and navigate to the DELAYED page, you can observe that the second marble view 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 subscribe to, at some point (think of broadcasts received in a BroadcastReceiver). 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 emit events only when subscribed to, but hot Observables emit events independent of being subscribed to.

Delayed elements

The other kind of delay in RxJava 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 DelayActivity. Replace the delayed subscription (that you just added) with:

sourceObservable
  .delay(delayInSeconds, TimeUnit.SECONDS,
    AndroidSchedulers.mainThread())
  .subscribe(delayed)

As you can see, the code is similar. You just replaced delaySubscription with delay. Run the app and look at the marble views. Can you spot the difference?

In the previous example, delaying the subscription made you miss the first three elements from the source Observable. When using the delay 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. Android comes with a few methods to accomplish timing tasks. Typically, Android developers use the Handler class to accomplish this sort of task. Handler works OK, but the API is somewhat complicated unless you wrap it, like we did in this app with the dispatchAfter function.

RxJava 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 the timer function several times to create interval timers through a handy custom function. In fact, the timer function uses another special RxJava function to achieve its timing tasks. Specifically, it uses the Observable.interval function. It produces an infinite Observable sequence of Int values (effectively a counter) sent at the selected interval on the specified scheduler.

In order to get some practice with the Observable.interval function, you’re going to go back through some of the work you did previously and replace instances of the timer function with a direct call to Observable.timer. Go back to ReplayActivity.kt class. Towards the beginning of the code, you created a source Observable. You used timer to create a timer and feed observers with values.

Delete the declaration of sourceObservable (including the replay()) and replace it with this instead:

val sourceObservable = Observable.interval(1L / elementsPerSecond,
  TimeUnit.SECONDS,
  AndroidSchedulers.mainThread()).replay(replayedElements)

And. That’s. All.

Interval timers are incredibly easy to create with RxJava. Not only that, but they are also easy to cancel: Since Observable.interval generates an Observable sequence, subscriptions 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, not immediately. 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 marble view if you run the app, values emitted by Observable.interval are integers starting from 0. Should you need different values, you can simply map them, or use the Observable.intervalRange function, which allows you supply both a starting value and a total number of items to emit. In most real life use-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 operator that is very much like Observable.interval 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. Open the DelayActivity.kt again. Locate the place where you used the delay operator. Replace the whole block of code with:

Observable.timer(3, TimeUnit.SECONDS)
  .flatMap {
    sourceObservable.delay(delayInSeconds, TimeUnit.SECONDS)
  }
  .subscribe(delayed)

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

• The whole chain is more readable (more “Rx-y”).
• Since the subscription returns a Disposable, you can cancel 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 Handler lambdas.

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 TimeoutException error event; if not caught, it terminates the sequence.

Open the TimeoutActivity.kt file. The associated activity_timeout.xml layout file contains a single MarbleView and a Button.

You’re going to use an extension from RxBindings that turns button taps into an Observable sequence. You’ll learn more about RxBindings 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.

In onCreate(), set up the Observable and connect it to the marble view:

button.clicks()
  .map { "•" }
  .timeout(5, TimeUnit.SECONDS)
  .subscribe(timeout)

Build and run, and click the “Timeout” button on the landing page. If you click the button within five seconds (and within five seconds of subsequent presses), you’ll see your taps on the marble view. Stop clicking, and five seconds after that, the timeout fires! The marble view will stop with an error donated by a big E.

An alternate version of timeout 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) then complete normally.

To try this, change the timeout call to the following:

.timeout(5, TimeUnit.SECONDS, Observable.just("X"))

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 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 marble view in a spot where only a transformation should occur.

Your task is to find an alternate way to do this. You can consider several approaches; try to 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.

1. The first one prepares the marble views (remember that side effects can be performed with the doOnNext operator).
2. The second one takes both the produced marble view and the source sequence element to generate a contextual value, every time window emits a new sequence. You might want to use a combination of zip and flatMap for this.

Key points

• 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.
• Buffering operators are a group of time-based operators that 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.
• dispatchAfter is a special function to make it easier to dispatch one-off actions. This displays elements received by the second subscription in another marble view.
• delaySubscription operators delay the time a subscriber starts receiving elements from its subscription. delay operators push the elements to they arrive later.
• The Observable.interval function produces an infinite Observable sequence of Int values (effectively a counter) sent at the selected interval on the specified scheduler.
• Timeout is an operator that semantically distinguishes an actual timer from a timeout (error) condition. Therefore, when a timeout operator fires, it emits an TimeoutException error event; if not caught, it terminates the sequence.