package scalabook.rx.chap1 
 
import rx.lang.scala.Observable 
import scala.concurrent.duration._ 
 
object SimpleRX extends App { 
   
  val o = Observable. 
            interval(100 millis). 
            take(5)             
             
  o.subscribe( x => println(s"Got it: $x") )             
  Thread.sleep(1000)           
             
  Observable. 
      just(1, 2, 3, 4). 
      reduce(_+_). 
      subscribe( r => println(s"Sum 1,2,3,4 is $r in a Rx Way")) 
   
} 

If you run this preceding Scala program, you will see the following output:

Got it: 0 
Got it: 1 
Got it: 2 
Got it: 3 
Got it: 4 
Sum 1,2,3,4 is 10 in a Rx Way 

If you try to run this code in the Scala REPL, it will fail, because we need the RxScala and RxJava dependencies. For this, we will need SBT and dependency management. Do not worry, we will cover how to work with SBT in our Scala application in the next chapter.

Going back to the observables, we need to import the Scala Observable. Make sure you get it from the Scala package, because if you get the Java one, you will have issues: in the very first part of the code, we will get numbers starting from 0 each 100 milliseconds, and this code would run forever. To avoid this, we use the take function to put a limit into the collection, so we will get the first five values. Then, later, we subscribe to the observer, and when data is ready, our code will run. For the first sample, it's pretty easy, we are just printing the values we have got. There is a thread sleep in this program, otherwise, the program would terminate, and you would not see any value on the console.

The second part of the code does something more interesting. First of all, it creates an Observable from a static list of values, which are 1,2,3, and 4. We apply a reduce function into the elements, which will sum all the elements with each other, and then we subscribe and print the result.

package scalabook.rx.chap1 
 
import rx.lang.scala.Observable 
 
object ComplexRxScala extends App { 
   
  Observable. 
      just(1,2,3,4,5,6,7,8,9,10).       // 1,2,3,4,5,6,7,8,9,10 
      filter( x => x%2==0).             // 2,4,6,8,10 
      take(2).                          // 2,4 
      reduce(_+_).                      // 6 
      subscribe( r => println(s"#1 $r")) 
   
   val o1 = Observable. 
            just(1,2,3,4,5,6,7,8,9,10).  // 1,2,3,4,5,6,7,8,9,10 
            filter( x => x%2==0).        // 2, 4, 6, 8, 10 
            take(3).                     // 2, 4 ,6     
            map( n => n * n)             // 4, 16, 36 
   
   val o2 = Observable.                  
            just(1,2,3,4,5,6,7,8,9,10). // 1,2,3,4,5,6,7,8,9,10  
            filter( x => x%2!=0).       // 1, 3, 5, 7, 9    
            take(3).                    // 1, 3, 5 
            map( n => n * n)            // 1, 9, 25 
            
   val o3 = o1. 
           merge(o2).                  // 2,16, 36, 1, 9, 25 
           subscribe( r => println(s"#2 $r")) 
       
} 

The preceding first part of the code creates an Observable with numbers from 1 to 10, and then applies a filter function, which will get only the even numbers. It then reduces them, calculates their sum, and lastly, prints the solution. You can visualize it as depicted in the following image:

For the second part of the code, we create two different observables. The first one is with even numbers and the second one is with odd numbers. These two observables are decoupled from each other; you can control as many observables you want. Later on, the code uses a merge function to join these two observables into a third and new observable containing the content of the first and second observables.

Merging 2 Observables

There are many functions and options, and you can see the whole list at http://rxmarbles.com/ and https://github.com/ReactiveX/RxScala. For the sake of simplicity, for now, we are just working with numbers. Later, we will use this to do more advance compositions including database calls and external web services calls.