So, let's start with an event with the user input. Go through the following example:

    fun main(args: Array<String>) { 
      println("Initial Out put with a = 15, b = 10") 
      var calculator:ReactiveCalculator = ReactiveCalculator(15,10) 
      println("Enter a = <number> or b = <number> in separate
      lines\nexit to exit the program") 
      var line:String? 
      do { 
        line = readLine(); 
        calculator.handleInput(line) 
      } while (line!= null && !line.toLowerCase().contains("exit")) 
    } 

If you run the code, you'll get the following output:

In the main method, we are not doing much operation except for just listening to the input and passing it to the ReactiveCalculator class, and doing all other operations in the class itself, thus it is modular. In the later chapters, we will create a separate observable for the input process, and we will process all user inputs there. We have followed the pull mechanism on the user input for the sake of simplicity, which you will learn to remove in the next chapters. So, let's now take a look at the following ReactiveCalculator class:

    class ReactiveCalculator(a:Int, b:Int) { 
      internal val subjectAdd: Subject<Pair<Int,Int>> = 
        PublishSubject.create() 
      internal val subjectSub: Subject<Pair<Int,Int>> =
        PublishSubject.create() 
      internal val subjectMult: Subject<Pair<Int,Int>> =
        PublishSubject.create() 
      internal val subjectDiv: Subject<Pair<Int,Int>> =
        PublishSubject.create() 
 
      internal val subjectCalc:Subject<ReactiveCalculator> =
        PublishSubject.create() 
 
      internal var nums:Pair<Int,Int> = Pair(0,0) 
 
      init{ 
        nums = Pair(a,b) 
 
        subjectAdd.map({ it.first+it.second }).subscribe
        ({println("Add = $it")} ) 
        subjectSub.map({ it.first-it.second }).subscribe
        ({println("Substract = $it")} ) 
        subjectMult.map({ it.first*it.second }).subscribe
        ({println("Multiply = $it")} ) 
        subjectDiv.map({ it.first/(it.second*1.0) }).subscribe
        ({println("Divide = $it")} ) 
 
        subjectCalc.subscribe({ 
          with(it) { 
            calculateAddition() 
            calculateSubstraction() 
            calculateMultiplication() 
            calculateDivision() 
          } 
         }) 
 
         subjectCalc.onNext(this) 
        } 
 
        fun calculateAddition() { 
          subjectAdd.onNext(nums) 
        } 
 
        fun calculateSubstraction() { 
          subjectSub.onNext(nums) 
        } 
 
        fun calculateMultiplication() { 
          subjectMult.onNext(nums) 
        } 
 
        fun calculateDivision() { 
          subjectDiv.onNext(nums) 
        } 
 
        fun modifyNumbers (a:Int = nums.first, b: Int = nums.second) { 
          nums = Pair(a,b) 
          subjectCalc.onNext(this) 
       } 
 
       fun handleInput(inputLine:String?) { 
        if(!inputLine.equals("exit")) { 
            val pattern: Pattern = Pattern.compile
            ("([a|b])(?:\\s)?=(?:\\s)?(\\d*)"); 
 
            var a: Int? = null 
            var b: Int? = null 
 
            val matcher: Matcher = pattern.matcher(inputLine) 
 
            if (matcher.matches() && matcher.group(1) != null 
            &&  matcher.group(2) != null) { 
              if(matcher.group(1).toLowerCase().equals("a")){ 
                 a = matcher.group(2).toInt() 
              } else if(matcher.group(1).toLowerCase().equals("b")){ 
                 b = matcher.group(2).toInt() 
               } 
            } 
 
            when { 
              a != null && b != null -> modifyNumbers(a, b) 
              a != null -> modifyNumbers(a = a) 
              b != null -> modifyNumbers(b = b) 
              else -> println("Invalid Input") 
           } 
        } 
      } 
    }      

In this program, we have push mechanism (observable pattern) only to the data, not the event (user input). While the initial chapters in this book will show you how to observe on data changes; RxJava also allows you to observer events (such as user input), we will get them covered during the end of the book while discussing RxJava on Android. So, now, let's understand how this code works.

First, we created a ReactiveCalculator class, which observes on its data and even on itself; so, whenever its property is modified, it calls all its calculate methods.

We used Pair to pair two variables and created four subject on the Pair to observe changes on it and then process it; we need four subject as there are four separate operations. You will also learn to optimize it with just one method in the later chapters.

On the calculate methods, we are just notifying the subject to process the Pair and print the new result.

If you focus on the map methods in both the programs, then you will learn that the map method takes the value that we passed with onNext and processes it to come up with a resultant value; that resultant value can be of any data type, and this resultant value is passed to the subscriber to process further and/or show the output.