Reactive Programming in Kotlin

Reactive programming is an asynchronous programming paradigm that revolves around data streams and the propagation of change. In simpler words, those programs which propagate all the changes that affected its data/data streams to all the interested parties (such as end users, components and sub-parts, and other programs that are somehow related) are called reactive programs.

For example, take any spreadsheet (say the Google Sheet), put any number in the A1 cell, and in the B1 cell, write the =ISEVEN(A1) function; it'll show TRUEor FALSE, depending on whether you've entered an even or odd number. Now, if you modify the number in A1, the value of B1 will also get changed automatically; such behavior is called reactive.

Not clear enough? Let's look at a coding example and then try to understand it again. The following is a normal Kotlin code block to determine if a number is even or odd:

    fun main(args: Array<String>) { 
      var number = 4 
      var isEven = isEven(number) 
      println("The number is " + (if (isEven) "Even" else "Odd")) 
      number = 9 
      println("The number is " + (if (isEven) "Even" else "Odd")) 
    } 
 
    fun isEven(n:Int):Boolean = ((n % 2) == 0) 

If you check the output of the program, then you'll see that, although the number is assigned a new value, isEven is still true; however, if isEven was made to track changes of the number, then it would automatically become false. A reactive program would just do the same.

So, let's first discuss the reasons to adapt functional reactive programming. There's no point in changing the whole way you code unless it gets you some really significant benefits, right? Yes, functional reactive programming gets you a set of mind-blowing benefits, as listed here:

So, what is the Reactive Manifesto? The Reactive Manifesto (http://www.reactivemanifesto.org) is a document defining the four reactive principles. You can think of it as the map to the treasure of reactive programming, or like the bible for the programmers of the reactive programming religion.

Everyone starting with reactive programming should have a read of the manifesto to understand what reactive programming is all about and what its principles are.

So, the following is the gist of four principles that Reactive Manifesto defines:

By implementing all four preceding principles, the system becomes reliable and responsive thus, reactive.

RxKotlin is a specific implementation of reactive programming for Kotlin, which is influenced by functional programming. It favors function composition, avoidance of global state, and side effects. It relies on the observer pattern of producer/consumer, with a lot of operators that allow composing, scheduling, throttling, transforming, error handling, and lifecycle management.

Whereas Reactor-Kotlin is also based on functional programming, and it is widely accepted and backed by the Spring Framework.

$ git clone https://github.com/ReactiveX/RxKotlin.git$ cd RxKotlin/$ ./gradlew build

compile 'io.reactivex.rxjava2:rxkotlin:2.x.y'

    <dependency> 
      <groupId>io.reactivex.rxjava2</groupId> 
      <artifactId>rxkotlin</artifactId> 
      <version>2.x.y</version> 
    </dependency> 

Comparing the pull mechanism with the RxJava push mechanism

RxKotlin revolves around the observable type that represents a system of data/events intended for push mechanism (instead of the pull mechanism of the iterator pattern of traditional programs), thus it is lazy and can be used synchronously and asynchronously.

It will be easier for us to understand if we start with a simple example that works with a list of data. So, here is the code:

    fun main(args: Array<String>) { 
      var list:List<Any> = listOf("One", 2, "Three", "Four", 4.5,
      "Five", 6.0f) // 1 
      var iterator = list.iterator() // 2 
      while (iterator.hasNext()) { // 3 
        println(iterator.next()) // Prints each element 4 
      } 
    } 

The following screenshot is the output:

So, let's go through the program line by line to understand how it works.

At comment 1, we're creating a list of seven items (the list contains data of mixed data types with the help of any class). On comment 2, we are creating iterator from the list, so that we can iterate over the data. In comment 3, we have created a while loop to pull data from the list with the help of iterator, and then, in 4, we're printing it.

The thing to notice is that we're pulling data from the list while the current thread is blocked until the data is received and ready. For example, think of getting that data from a network call/database query instead of just List and, in that case, how long the thread will be blocked. You can obviously create a separate thread for those operations, but then also, it will increase complexity.

Just give a thought; which one is a better approach? Making the program wait for data or pushing data to the program whenever it's available?

The building blocks of the ReactiveX Framework (be it RxKotlin or RxJava) are the observables. The observable class is just the opposite ofiterator interface. It has an underlying collection or computation that produces values that can be consumed by a consumer. However, the difference is that the consumer doesn't pull these values from the producer, like in the iterator pattern; instead, the producer pushes the values as notifications to the consumer.

So, let's take the same example again, this time with observable:

    fun main(args: Array<String>) { 
      var list:List<Any> = listOf("One", 2, "Three",
      "Four", 4.5, "Five", 6.0f) // 1 
      var observable: Observable<Any> = list.toObservable(); 
 
       observable.subscribeBy( // named arguments for
       lambda Subscribers 
         onNext = { println(it) }, 
         onError =  { it.printStackTrace() }, 
         onComplete = { println("Done!") } 
      ) 
    } 

This program output is the same as the previous one—it prints all the items in the list. The difference is in the approach. So, let's see how it actually works:

1. Create a list (just the same as the previous one).
2. An observable instance is created with that list.
3. We're subscribing to the observer instance (we're using named arguments for lambda and covering it in detail later).

As we subscribe to observable, each data will be pushed to onNext, and, as it gets ready, it will call onComplete when all data is pushed and onError if any error occurs.

So, you learned to use the observable instances, and they are quite similar to the iterator instances, which is something we're very familiar with. We can use these observable instances to build asynchronous streams and push data updates to their subscribers (even to multiple subscribers).This was a simple implementation of the reactive programming paradigm. The data is being propagated to all the interested parties—the subscribers.

The ReactiveEvenOdd program

So, now that we are somewhat familiar with observables, let's modify the even-odd program in a reactive way. Here is the code for doing so:

    fun main(args: Array<String>) { 
      var subject:Subject<Int> = PublishSubject.create() 
 
      subject.map({ isEven(it) }).subscribe({println
      ("The number is ${(if (it) "Even" else "Odd")}" )}) 
 
      subject.onNext(4) 
      subject.onNext(9) 
    } 

Here is the output:

In this program, we have used subject and map, which we will cover in the later chapters. Here, it is just to show how easy it is in reactive programming to notify the changes. If you look at the program closely, then you'll also find that the code is modular and functional. When we notify subject with a number, it calls the method in map, then it calls the method in subscribe with the return value of the map method. The map method checks if the number is even and returns true or false accordingly; in the subscribe method, we are receiving that value and printing even or odd accordingly. The subject.onNext method is the way through which we message the new value to the subject, so it can process it.

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.

In this chapter, we learned about what reactive programming is and the reasons we should learn it. We also started with coding. The reactive coding pattern may seem new or somehow uncommon, but it is not that hard; while using it, you just need to declare a few more things.

We learned about observable and its use. We also got introduced to subject and map, which we will learn in depth in the later chapters.

We will continue with ReactiveCalculator example in the later chapters and see how we can optimize and enhance this program.

The three examples presented in this chapter may seem a bit confusing and complex at first, but they're really simple, and they will become familiar to you as you proceed with this book.

In the next chapter, we will learn more about functional programming and functional interfaces in RxKotlin.