Rx for the Ultimate Beginner – Part 2 (Kotlin)

|by

This is the second part of our ongoing “Rx for the Ultimate Beginner” series. If you haven’t read the first one, I strongly suggest you do so. As always, this post has a twin sibling for iOS developers with code examples given in Swift.

Observables Explained

Let’s recap what we did before to fully understand the potential observables promise.

fun observableExample() {
    val myObservable = Observable.fromArray(1, 2, 3)
    val transformation = myObservable.map { e ->
        //Log.d("Rx","Map: " + e)
        e * 2
    }
    val filtered = transformation.filter { e ->
        //Log.d("Rx","Filter: " + e)
        e > 2
    }

    filtered.subscribe(
        { e ->
            // onNext
            Log.d("Rx","Result: " + e)
        },
        { ex ->
            // onError
            Log.e("Rx", "Error", ex)
        },
        {
            // onComplete
            Log.d("Rx","Done!")
        },
        { disposable ->
            //onSubscribe
            Log.d("Rx","Subscribed!")
        }
    )
}

// Output is
// Result: 4
// Result: 6
// Done!
// Garbage collected!

As you can see, we created an observable with the operator Observable.fromArray(). Then, we chained a bunch of transformation one after another. Finally, we subscribed to our final observable to receive the calculated results.

The first thing to note here is that all the transformation functions we called there (map and filter in our case) also return observables of type Observable<Int>. This is one of the main benefits of the functional paradigm. All results are explicitly available to the caller in case he/she needs to reuse them in other contexts. Moreover, having them as observables allows the developer to subscribe to any intermediate result independently of the chain it belongs to. In other words, we can build a railway network of forking and joining transformations wherein the end multiple subscriptions act on different results according to the recipe.

Observables allow such behavior by ensuring two guarantees in their protocol. First, the pipeline does not execute before a subscriber subscribes. Second, if multiple subscribers subscribe, unless told explicitly, a copy of the chain is run for each subscriber independently. In other words, when no one subscribes, the chain never runs. In this regard, the chain itself acts like some kind of type where the subscriptions are its instances. This behavior is this way by design. Observables are meant to be declared and stored for multiple uses. What flows through the pipe is only the subscriber’s concern.

In the future, we will learn how to create our own observables. With that approach, we will be able to, for instance, initiate a chain with an HTTP call. Before we dive into how to do it, just assume the first line here represents an API call. Then ask yourself what happens if you subscribe to it multiple times.

let apiResult:Observable<JSON> = // some HTTP call wrapper for Rx

_ = api.subscribe()
_ = api.subscribe()

As you may have guessed, this would make the call happen twice. This is also the case if we made the call in the middle of a chain instead of in the beginning. Observable’s being nothing but recipes is, therefore, their strongest feature.

Subscription Lifecycle

When we subscribe to an observable, we provide 4 callbacks, each notified on different events. onNext is called every time a new emission is made. onCompleted is called right after the last emission is received. onError is called if anything bad happens during the chain’s execution. onDisposed is called when the subscription ends and goes out of scope.

The observable protocol ensures that onNext is expected to be called 0, 1 or any number of times. Not a single onNext can be called after an onError or onCompleted happens. If onError is called, onCompleted or onNext cannot be called anymore and the subscription terminates. Likewise, if onCompleted is called, onError or onNext cannot be called anymore and the subscription terminates.

In the end, the lifecycle of a subscription looks like a bunch of onNext calls potentially ending with a single onCompleted or onError. Since there is no limit for the number of onNext occurences, there can be subscriptions that practically never end. For example, a gyro sensor in a device can potentially provide an infinite amount of data to its subscribers. In such cases onCompleted has no purpose.

Traditionally, to better communicate with our fellow developers, observables can be categorized by the number of times or frequency onNext is expected to be called. They also differ in the way their subscribers affect the environment. Daniel Lew explains the differences in one of his great talks.

Cold Observables vs Hot Observables

An Observable is considered hot if:

  • The source produces sequences even if no subscriber is connected.
  • Subscribers do not cause side effects on the source.

An observable is considered cold if:

  • The source only produces sequences when a subscriber connects.
  • Subscribers cause side effects on the source.

By this description, it is clear that sources like user input and device sensors can be considered candidates of hot observables. The number of times we subscribe cannot affect the moment a user click event or a device sensor reading (gyro, GPS etc) occurs. Even if there is no subscriber, a user can still touch a button on the UI despite causing no effect.

On the other hand, sources who communicate through network or file system are good candidates for cold observables. Reading from a file can emit different sequences for different subscribers if they do not read with the same pace. An HTTP call can cause a database write, hence affecting the behavior of the next call another subscriber makes (i.e it may fail if duplicate writes are forbidden).

Conclusion

We learned how an observable notifies its subscribers and in what sense can we categorize observables. The Observable Contract is a pretty good read if you want to fully grasp the formalism behind observables’ behavior. Next time, we are gonna create our own observables and glue them together to fix real-life problems.

Have a nice day!

References