Learning RxJava: Thoughts on Chapter 12

Kotlin

Kotlin can work extremely well with RxJava. In general, Kotlin is a cleaner and simpler language than Java, but is still compatible with Java code and Java libraries. Though it should be noted that there can sometimes be some compatibility issues with certain Java libraries and SAM ambiguity. JetBrains is aware of this issue and is looking into a fix.

In the meantime, there is a helper library called RxKotlin that works around these issues with the RxJava library, in addition to giving a handful of tools.

One big plus with using Kotlin is that is supports extension functions. This in turn allows for a much cleaner implementation of custom operators, since they can be called directly from the previous operator instead of having to call compose() or lift() and passing in the custom operator.

There has been talk in the Kotlin community about implementing a pure Kotlin Rx library. With a quick look around the internet, this looks like it could be promising, though it's still a work in progress: https://github.com/badoo/Reaktive

Learning RxJava: Thoughts on Chapter 11

RxJava on Android

In short, Rx is extremely popular on mobile platforms, with RxJava being heavily used on Android and RxSwift being used on iOS.

One of the pain points, though, is that a large number of Android devices are still stuck with Java 6, which is missing lambda functions. The same thing can still be accomplished with anonymous classes, but they are extremely verbose. To fix this issue, you can use Retrolambda, which will allow you to use Java 8 style lambdas, but will then compile it to use anonymous classes in the byte code. Another option is just to forego Java altogether, and use Kotlin instead, which many developers have opted to do.

Two important libraries to include with android development are RxAndroid and RxBinding. The RxAndroid library has a handful of useful tools, primary of which are the Android Schedulers. With this you can switch between the main thread for UI elements and other threads for processing. RxBinding is an Rx wrapper for UI elements. There are also many other libraries available that can be useful depending on your use case. Go to https://github.com/ReactiveX/RxAndroid/wiki to see a list of these libraries.

Learning RxJava: Thoughts on Chapter 10

Testing and Debugging

Blocking Subscribers and Operators

There are blocking subscribers and operators that can be used to ensure the test waits until you get your results from the chain so that you can run your tests against them. While these can be useful in tests, frequently the TestObserver and TestSubscriber are better choices. Note that while it can be tempting to use blocking subscribers and operators in production code, you should avoid doing so, because it limits the effectiveness and flexibility of your reactive code.

TestObserver and TestSubscriber

TestObserver (for Observables) and TestSubscriber (for Flowables) make for very powerful and flexible tools for writing unit tests. They will keep track of data related to emissions, errors, and completion events, which allows you to examine and assert this data afterwards. They even have assertion methods built into them, such as assertValueCount, assertValues, etc.

TestScheduler

The TestScheduler is a special scheduler that allows you to manipulate time for testing purposes. It has methods such as advanceTimeBy and advanceTimeTo that will allow you to test time bound code quickly. The book does note that the TestScheduler  "is not a thread-safe Scheduler and should not be used with actual concurrency." If the code that you wish to test doesn't allow easy access to setting the Scheduler, you can instead use RxJavaPlugins.setComputationScheduler() or other such methods that will override the standard Schedulers and inject the TestScheduler in its place.

Debugging Strategies

Debugging Rx code can be difficult, since frequently the stack traces can be less than useful and it can sometimes be difficult to apply breakpoints where you need them, but a strategy that can work well is to take advantage of the doOnNext operator and other similar operators to be able to see what's going on at every step of the chain and to track down where the faulty link in the chain is located.

Learning RxJava: Thoughts on Chapter 9

compose()

You can create custom operators through composing existing operators by implementing the ObservableTransformer interface, and then passing it in to the compose() operator. This is useful for extracting duplicated code.

Often, creating a static method that will return the ObservableTransformer will allow for a clean and easy way to pass it into the compose operator.

This can also be done for Flowables using the FlowableTransformer.

to()

The to() operator simply takes a Function<Observable<T>, R>, which can be used to fluently convert the Observable to something else. This can be quite useful for mapping from an Observable to a language or framework specific construct such as a Future.

lift()

Sometimes there arises the need to build a custom operator from scratch, instead of composing it from existing operators. In such a situation you can implement the ObservableOperator interface and pass the resulting object to the lift() operator.

Note that this should be a rare occurrence, and it can be quite difficult to get right. Try your best to first implement your needs using the ObservableTransformer and the compose() operator, but if you determine that this is insufficient, then take a close look at how the standard operators are implemented in RxJava source code as well as trusted operator libraries such as RxJava2-Extras.

Learning RxJava: Thoughts on Chapter 8

Backpressure and Flowables

Backpressure is only needed in multi-threading situations with a large number of emissions. When the rx call chain is all executed on one thread, the thread will take each emission one at a time all the way through the chain from the source observable to the final observer. When the chain is executed across multiple threads, then the first thread will take an emission through the chain to the hand off point and hand it off to the next thread, and then immediately get the next emission and move it through.

If the second thread is slower at moving emissions through its part of the call chain, then the number of emissions that the first thread has finished but the second thread has yet to start can pile up. If the total number of emissions is small, then this isn't really an issue, but with a large number of emissions this pile up could potentially lead to an out of memory error. This is where backpressure and Flowables come in.

Essentially, using a Flowable in this example is telling thread one to slow down. It would do this by having thread one process a certain number of emissions at the start, and then wait until thread two has pulled through a certain percentage of those emissions before processing more. In this way, it tries to keep a small buffer of emissions between the two threads without letting it get out of control, thus keeping thread 2 busy without pause, but without overloading memory with thread one going nonstop.

Note that Flowable does add overhead, so opt for Observable in cases where backpressure isn't needed, but when it is needed, simply switching Observable to Flowable should be sufficient in most cases. Though there are a handful of exceptions.

Since dealing with these exceptions is infrequent, I don't feel the need to go into detail here, but if you do run into such a situation, there are a number of options available, depending on the situation, varying from using an onBackpressureXXX() operator to creating your own custom Flowable.

Learning RxJava: Thoughts on Chapter 7

Buffering

The buffer() operator will gather emissions into collections according to a specified criteria. The default collection type is as a list, but a different collection type can be specified.

You can specify a count, and the buffer operator will group emissions into lists with a size equal to the count, with the exception of the last list, which will contain the remainder of what couldn't be divided equally.

You can additionally specify a skip amount as well, which determines how far to move forward for the start of each list. If no skip is given it defaults to be the same as the count operator so as to break them into distinct groups. But if, for instance, you specify a count of 2 and a skip of 1, then each list will have 2 elements, but the start of a list will only be one more than the start of the previous list, which will cause the last element of the previous list the first element of the current list to be the same. This can be very useful for operations where you need to know both the current emission and the previous emission to do something.

You can also buffer based off of time, so that all emissions within a specified time are grouped together. There is also a timeSkip option, which is the time based equivalent of skip. There is also an optional count operation, so that it will group based off of whichever is reached first, the time or the count.

Additionally, buffer can take another Observable as a parameter, and anytime that Observable emits serves as the cutoff point for grouping.

Windowing

Windowing works the exact same as buffering, but instead of grouping the emissions into collections, the emissions are grouped into Observables. This can be useful in that it will allow you to work with the emissions coming in immediately, instead of waiting for the last one to be available before you can look at any of them.

Throttling

Throttling will throw away emissions when they are coming too fast. Variants include throttleLast(), throttleFirst(), and throttleWithTimeout(). throttleLast()/sample() will only emit the last item from a fixed time interval. throttleFirst() will only emit the first item from a fixed time interval. throttleWithTimeout()/debounce() will wait until there is a pause of a specified length, and then it will send the last emission from before the pause. The downside is that emissions are delayed until the end of the specified time period.

Switching

switchMap() is similar to flatMap, in that it maps an emission to an observable. The difference between them being that while flatMap will combine the emissions from all Observables into a single Observable emiting all emissions, switchMap will unsubscribe from an Observable as soon as an emission comes in and creates a new Observable. So at any given point, it is only passing on the emissions of one Observable, and that one Observable is the one created by the most recently received emission. This can be useful in cancelling and restarting expensive operations that are kicked off by user events.

Learning RxJava: Thoughts on Chapter 6

Concurrency and Parallelization

Simply put, in order to utilize the full power and speed of the CPU, you need to be able to run things concurrently. Concurrency (also called multithreading) is essentially multitasking, or performing more than one thing at the same time. The modern CPU has multiple cores, and the only way to utilize multiple cores is through multithreading. Without multithreading, you are limited to a single core.

There are a number of gotchas around concurrency, which RxJava tries to smooth out and make simple. For instance, there's a fair amount of overhead in creating a thread, so a best practice in multithreading is instead to have a thread pool, where threads are previously created, and when a thread is needed, it is pulled out of this pool, and when it is no longer needed, it is returned to this pool instead of being destroyed. RxJava strives to make this process simple through the use of Schedulers.

Schedulers

Schedulers are essentially predefined thread pool managers, with there being multiple different managers for the different types of tasks to be performed.

Computation

The computation scheduler is designed around computation heavy tasks; the kind of tasks that would require the full use of the CPU core. These tasks usually focus around math, algorithms, or complex logic. To handle this, the computation scheduler limits the number of threads in its thread pool based off of the processor count available to the JVM.

IO

The io scheduler is generally used for waiting on devices or protocols that are slow, where a fair amount of what the thread needs to do is sit around and wait for a response. This is commonly the case with disk read operations or calls over a network. To handle these, the io scheduler will try to match the number of threads in it's pool to the number of tasks that are needing a thread. To do this, it will dynamically grow or shrink its thread pool.

New Thread

The new thread scheduler is straightforward enough. It will create a new thread for each Observer, and then destroy the thread when it is done. There is no thread pool. This is useful for the every once in a while situation where it makes sense to have an individual thread for a very specific purpose.

Single

The single scheduler has a single thread in its pool. This can be useful "to isolate fragile, non-threadsafe operations to a single thread."

Trampoline

"In practicality, you will not invoke [the trampoline scheduler] often as it is used primarily in RxJava's internal implementation. [...] It is just like the default scheduler on the immediate thread, but it prevents cases of recursive scheduling where a task schedules a task while on the same thread. Instead of causing a stack overflow error, it will allow the current task to finish and then execute that new scheduled task afterward."

ExecutorService

You can build a custom scheduler off of an ExecutorService. This will allow you to have fine grained control over the thread pool and the rules used to govern it. This is useful for those circumstances where the defaults don't fit your needs.

Starting and Stopping Schedulers

Each of the default schedulers is lazily instantiated. At any point in time you can call shutdown() on any of them to immediately stop its threads, or call Schedulers.shutdown() to stop them all at once. After that, you can call start() to start any of them back up, or Schedulers.start() to start them all back up at once.

Using Schedulers

To use schedulers, you will utilize the subscribeOn(), observeOn(), and unsubscribeOn() methods.

subscribeOn()

The subscribeOn() method is used to suggest which scheduler should be used to begin the Observable chain. The placement of the subscribeOn() method in the chain has no effect, since in each case it will suggest that the Observable chain be started off on the specified scheduler, but in general best practice is to keep it as close to the source Observable as possible. Note that subscribeOn() will not work with certain Observable factories, such as Obsevable.interval. In such cases, these factories will have a method overload that will allow you to specify the scheduler directly to the factory.

observeOn()

The observeOn() method is used to switch to a different scheduler at that point in the Observable chain, such that the part above the observeOn() call will be run on one thread, and the part below will be run on a different thread. The observeOn() serves as a bridge, moving data from one thread to the next.

This can be particularly useful in applications that have a dedicated UI thread, allowing you to switch off of the UI thread to perform computations, then switch back to the UI thread to update the user interface. In this was you can keep the UI from freezing.

unsubscribeOn()

If unsubscribing from an observable is particularly costly, the unsubscribeOn() method can be used to specify that the unsubscribe code be run on a different thread. This can be useful in situations where, for instance, database connections need to be closed on unsubscribe.

Parallelization with flatMap()

If you have a lot of emissions that you need to send through some intensive or time consuming operations and you'd like to speed up the process by performing these operations across multiple threads, you can do so by taking advantage of flatMap(). FlatMap merges multiple Observables together, and it's designed to work even if those Observables are on different threads.

So, to run these operations in parallel, take each emission (or group of emissions) and flatMap it to an Observable that uses either subscribeOn or observeOn to move it to another thread. It's as simple as that.

Major Takeaways

One of the things that makes Rx extremely powerful is the ability to take sequential code and in just one or two lines transform it into code that runs on a separate thread or runs mutiple threads in parallel. It greatly simplifies concurrency and parallelization.