ReactiveX

home polyglot operators code

RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.

It extends the observer pattern to support sequences of data/events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety, concurrent data structures, and non-blocking I/O.

It supports Java 5 or higher and JVM-based languages such as Groovy, Clojure, JRuby, Kotlin and Scala.

Observables fill the gap by being the ideal implementation of access to asynchronous sequences of multiple items
single itemsmultiple items
synchronousT getData()Iterable<T> getData()
asynchronousFuture<T> getData()Observable<T> getData()

Why?

Observables are Composable

Java Futures are straightforward to use for a single level of asynchronous execution but they start to add non-trivial complexity when they’re nested.

It is difficult to use Futures to optimally compose conditional asynchronous execution flows (or impossible, since latencies of each request vary at runtime). This can be done, of course, but it quickly becomes complicated (and thus error-prone) or it prematurely blocks on Future.get(), which eliminates the benefit of asynchronous execution.

RxJava Observables on the other hand are intended for composing flows and sequences of asynchronous data.

Observables are Flexible

RxJava’s Observables support not just the emission of single scalar values (as Futures do), but also of sequences of values or even infinite streams. Observable is a single abstraction that can be used for any of these use cases. An Observable has all of the flexibility and elegance associated with its mirror-image cousin the Iterable.

An Observable is the asynchronous/push "dual" to the synchronous/pull Iterable
eventIterable (pull)Observable (push)
retrieve dataT next()onNext(T)
discover errorthrows ExceptiononError(Exception)
completereturnsonCompleted()

Observables are Less Opinionated

The RxJava implementation is not biased toward some particular source of concurrency or asynchronicity. Observables in RxJava can be implemented using thread-pools, event loops, non-blocking I/O, actors (such as from Akka), or whatever implementation suits your needs, your style, or your expertise. Client code treats all of its interactions with Observables as asynchronous, whether your underlying implementation is blocking or non-blocking and however you choose to implement it.

RxJava also tries to be very lightweight. It is implemented as a single JAR that is focused on just the Observable abstraction and related higher-order functions. You could implement a composable Future that is similarly unbiased, but Akka Futures for example come tied in with an Actor library and a lot of other stuff.)

How is this Observable implemented?
public Observable getData();
From the Observer's point of view, it doesn't matter!
  • does it work sychronously on the same thread as the caller?
  • does it work asynchronously on a distinct thread?
  • does it divide its work over multiple threads that may return data to the caller in any order?
  • does it use an Actor (or multiple Actors) instead of a thread pool?
  • does it use NIO with an event-loop to do asynchronous network access?
  • does it use an event-loop to separate the work thread from the callback thread?

And importantly: with RxJava you can later change your mind, and radically change the underlying nature of your Observable implementation, without breaking the consumers of your Observable.

Callbacks Have Their Own Problems

Callbacks solve the problem of premature blocking on Future.get() by not allowing anything to block. They are naturally efficient because they execute when the response is ready.

But as with Futures, while callbacks are easy to use with a single level of asynchronous execution, with nested composition they become unwieldy.

RxJava is a Polyglot Implementation

RxJava is meant for a more polyglot environment than just Java/Scala, and it is being designed to respect the idioms of each JVM-based language. (This is something we’re still working on.)

Functional Reactive Programming (FRP)

RxJava provides a collection of operators with which you can filter, select, transform, combine, and compose Observables. This allows for efficient execution and composition.

You can think of the Observable class as a “push” equivalent to Iterable, which is a “pull.” With an Iterable, the consumer pulls values from the producer and the thread blocks until those values arrive. By contrast, with an Observable the producer pushes values to the consumer whenever values are available. This approach is more flexible, because values can arrive synchronously or asynchronously.

Example code showing how similar high-order functions can be applied to an Iterable and an Observable
IterableObservable
 
getDataFromLocalMemory()
  .skip(10)
  .take(5)
  .map({ s -> return s + " transformed" })
  .forEach({ println "next => " + it })
 
getDataFromNetwork()
  .skip(10)
  .take(5)
  .map({ s -> return s + " transformed" })
  .subscribe({ println "onNext => " + it })

The Observable type adds two missing semantics to the Gang of Four’s Observer pattern, to match those that are available in the Iterable type:

  1. the ability for the producer to signal to the consumer that there is no more data available (a foreach loop on an Iterable completes and returns normally in such a case; an Observable calls its observer's onCompleted() method)
  2. the ability for the producer to signal to the consumer that an error has occurred (an Iterable throws an exception if an error takes place during iteration; an Observable calls its observer's onError() method)

With these additions, RxJava harmonizes the Iterable and Observable types. The only difference between them is the direction in which the data flows. This is very important because now any operation you can perform on an Iterable, you can also perform on an Observable.

We call this approach Functional Reactive Programming because it applies functions (lambdas/closures) in a reactive (asynchronous/push) manner to asynchronous sequences of data. (This is not meant to be an implementation of the similar but more restrictive “functional reactive programming” model used in languages like Fran.)

More Information

RxJava Libraries

The following external libraries can work with RxJava:

Reactive Extensions
Composable Observables

github.com/ReactiveX
twitter.com/ReactiveX