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 var known as 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...

This chapter is all about Reactive Extensions, so let's go ahead and create a project called rx-playground which 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.9.0"] 
                 [io.reactivex/rxclojure "1.0.0"]]) 

Now, fire up a REPL in the project&apos...

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 one 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...

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

In this section, we will 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...

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 will examine here.

Let's get started by creating a badly behaved observable that always throws an exception:

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

Now, let's watch what happens if we subscribe to it:

(rx/subscribe...

Another issue we might be faced with is observables that produce items faster than we can consume them. The problem that arises in this scenario is to do with the 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...

In this chapter, we took a deep dive into RxJava, a port form of 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 of the concepts that have been presented are extremely useful in real applications and will come in handy for our next chapter, where we will 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.

In the next chapter we will create a stock market monitoring application using observable sequences and RxClojure. They will help us reduce the complexity...

Here is a list of information you can refer to regarding what we covered in this chapter:

1. Subject/Observer is Dual to Iterator, Erik Meijer:
   http://csl.stanford.edu/~christos/pldi2010.fit/meijer.duality.pdf