Custom observables

Rx provides many more factory methods to create observables, but it is beyond the scope of this book to cover them all. Interested readers are advised to consult the Rx documentation (see https://github.com/ReactiveX/RxJava/wiki/Creating-Observables) which provides examples of other factory methods.

Nevertheless, sometimes, none of the built-in factories are what you want. For such cases, Rx provides the create method. We can use it to create a custom observable from scratch.

As an example, we'll create our own version of the just observable we used earlier in this chapter:

(defn just-obs [v] 
  (rx/observable* 
   (fn [observer] 
     (rx/on-next observer v) 
     (rx/on-completed observer)))) 
 
(rx/subscribe (just-obs 20) prn) 

First, we create a function, just-obs, which implements our observable by calling the observable* function.

When creating an observable in this way, the function passed to observable* will get called with an observer as soon as one subscribes to us. When this happens, we are free to do whatever computation—and even I/O—we need in order to produce values and push them to the observer.

We should remember to call the observer's onCompleted method whenever we're done producing values. The preceding snippet will print 20 to the REPL.