Clojure Reactive Programming: Design and implement highly reusable reactive applications by integrating different frameworks with Clojure

In Chapter 1, What is Reactive Programming?, we saw a brief overview of the Observer design pattern and a simple implementation of it in Clojure using watches. Here's how we did it:

(def numbers (atom []))

(defn adder [key ref old-state new-state]
  (print "Current sum is " (reduce + new-state)))

(add-watch numbers :adder adder) 

In the preceding example, our observable subject is the var, numbers. The observer is the adder watch. When the observable changes, it pushes its changes to the observer synchronously.

Now, contrast this to working with sequences:

(->> [1 2 3 4 5 6]
     (map inc)
     (filter even?)
     (reduce +))

This time around, the vector is the subject being observed and the functions processing it can be thought of as the observers. However, this works in a pull-based model. The vector doesn't push any elements down the sequence. Instead, map and friends ask the sequence for more elements. This is a synchronous operation.

Rx makes sequences...

This chapter is all about Reactive Extensions, so let's go ahead and create a project called rx-playground that we will be using in our exploratory tour. We will use RxClojure (see https://github.com/ReactiveX/RxClojure), a library that provides Clojure bindings for RxJava() (see https://github.com/ReactiveX/RxJava):

$ lein new rx-playground

Open the project file and add a dependency on RxJava's Clojure bindings:

(defproject rx-playground "0.1.0-SNAPSHOT"
  :description "FIXME: write description"
  :url "http://example.com/FIXME"
  :license {:name "Eclipse Public License"
            :url "http://www.eclipse.org/legal/epl-v10.html"}
  :dependencies [[org.clojure/clojure "1.5.1"]
                 [io.reactivex/rxclojure "1.0.0"]])"]])

Now, fire up a REPL in the project's root directory so that we can start creating some observables:

$ lein repl

The first thing we need to do is import RxClojure, so let's get this out of the way by typing the following in the REPL:

(require...

Now that we know how to create observables, we should look at what kinds of interesting things we can do with them. In this section, we will see what it means to treat Observables as sequences.

We'll start with something simple. Let's print the sum of the first five positive even integers from an observable of all integers:

(rx/subscribe (->> (Observable/interval 1 TimeUnit/MICROSECONDS)
                   (rx/filter even?)
                   (rx/take 5)
                   (rx/reduce +))
                   prn-to-repl)

This is starting to look awfully familiar to us. We create an interval that will emit all positive integers starting at zero every 1 microsecond. Then, we filter all even numbers in this observable. Obviously, this is too big a list to handle, so we simply take the first five elements from it. Finally, we reduce the value using +. The result is 20.

To drive home the point that programming with observables really is just like operating on sequences...

In the previous section, we learned how to transform and combine observables with operations such as map, reduce, and zip. However, the two observables above—musicians and bands—were perfectly capable of producing values on their own. They did not need any extra input.

In this section, we examine a different scenario: we'll learn how we can combine observables, where the output of one is the input of another. We encountered flatmap before in Chapter 1, What is Reactive Programming? If you have been wondering what its role is, this section addresses exactly that.

Here's what we are going to do: given an observable representing a list of all positive integers, we'll calculate the factorial for all even numbers in that list. Since the list is too big, we'll take five items from it. The end result should be the factorials of 0, 2, 4, 6, and 8, respectively.

The first thing we need is a function to calculate the factorial of a number n, as well as our observable:

(defn factorial...

A very important aspect of building reliable applications is knowing what to do when things go wrong. It is naive to assume that the network is reliable, that hardware won't fail, or that we, as developers, won't make mistakes.

RxJava embraces this fact and provides a rich set of combinators to deal with failure, a few of which we examine here.

(defn exceptional-obs []
  (rx/observable*
   (fn [observer]
     (rx/on-next observer (throw (Exception. "Oops. Something went wrong")))
     (rx/on-completed observer))))

(rx/subscribe (->> (exceptional-obs)
                   (rx/map inc))
              (fn [v] (prn-to-repl "result is " v)))

;; Exception Oops. Something went wrong  rx-playground.core/exceptional-obs/fn--1505

Another issue we might be faced with is the one of observables that produce items faster than we can consume. The problem that arises in this scenario is what to do with this ever-growing backlog of items.

As an example, think about zipping two observables together. The zip operator (or map in RxClojure) will only emit a new value when all observables have emitted an item.

So if one of these observables is a lot faster at producing items than the others, map will need to buffer these items and wait for the others, which will most likely cause an error, as shown here:

(defn fast-producing-obs []
  (rx/map inc (Observable/interval 1 TimeUnit/MILLISECONDS)))

(defn slow-producing-obs []
  (rx/map inc (Observable/interval 500 TimeUnit/MILLISECONDS)))

(rx/subscribe (->> (rx/map vector
                           (fast-producing-obs)
                           (slow-producing-obs))
                   (rx/map (fn [[x y]]
                             (+ x y)))
                  ...

In this chapter, we took a deep dive into RxJava, a port form Microsoft's Reactive Extensions from .NET. We learned about its main abstraction, the observable, and how it relates to iterables.

We also learned how to create, manipulate, and combine observables in several ways. The examples shown here were contrived to keep things simple. Nevertheless, all concepts presented are extremely useful in real applications and will come in handy for our next chapter, where we put them to use in a more substantial example.

Finally, we finished by looking at error handling and backpressure, both of which are important characteristics of reliable applications that should always be kept in mind.