The TestObserver is an RxJava staple for testing.

It allows you to assert values in a stream, in the specific order they were emitted. Here’s a quick code snippet from the movies-usf repository ¹:

@Test
fun onSearchingForMovieBladeRunner_shouldSeeSearchResult() {
    viewModel = MSMainVm(mockApp, mockMovieRepo)
    val viewStateTester = viewModel.viewState.test()

    viewModel.processInput(SearchMovieEvent("blade runner 2049"))

    viewStateTester.assertValueAt(1) {
        assertThat(it.searchedMovieTitle).isEqualTo("Searching Movie...")
        true
    }

    viewStateTester.assertValueAt(2) {
        assertThat(it.searchedMovieTitle).isEqualTo("Blade Runner 2049")
        // ...
        true
    }
}

If you look at the source for the base TestObserver, there are a bunch of these useful methods:

A not so well known api in RxJava is the .hide() operator.

When does one use the hide operator in Rx?

From the docs:

Hides the identity of this Observable and its Disposable.

Allows hiding extra features such as Subject’s Observer methods or preventing certain identity-based optimizations (fusion). there are a lot of complex operations that take place internally in RxJava (like internal queue creation, worker instantiation + release, numerous atomic variables being created and modified.)

Ok, so in my previous post I innocuously introduced the .share() operator.

Observable<Object> tapEventEmitter = _rxBus.toObserverable().share();

What is this share operator?

The .share() operator is basically just a wrapper to the chained call .publish().refcount().

You’ll find the chained combo .publish().refcount() used in quite a few Rx examples on the web. It allows you to “share” the emission of the stream. Considering how powerpacked and frequently used this combo is, RxJava basically introduced the friendlier more useful operator share(). This mechanism is sometimes referred to as “multicasting”.

This is a bonus RxJava post that I landed up writing along with my previous post on creating an event bus with RxJava. If you went through the code in the actual repo you would have noticed more than one version of the bottom fragment in the RxBus demo.

Originally I envisioned the RxBus example being a tad bit fanicer however as I coded up the example, I realized that too many concepts were getting conflated. The ridiculous simplicity of the RxBus implementation was lost. So I dumbed down the original example but left in the original code for the Rx padawans.

This post has three parts:

1. quick primer on what an event bus is
2. implementing the event bus with RxJava
3. parting thoughts on this approach

“RxBus” is not going to be a library. Implementing an event bus with RxJava is so ridiculously easy that it doesn’t warrant the bloat of an independent library.

Part 1: What is an event bus?

Let’s talk about two concepts that seem similar: the Observer pattern and the Pub-sub pattern.

I’ve read and watched a lot on Rx. Most examples either use the J8 lambda notations/Scala/Groovy or some other awesome language that us Android developers are constantly envious of.

Unfortunately I could never find real-world simple examples in Android that could show me how to use RxJava. To scratch that itch, I created a github repo with a couple of nifty examples using RxJava. You can build the apk and run the app to see different working examples.