Reactive Programming in Kotlin

Along with the Reactive Manifesto, we also have a standard specification on Reactive Streams. Everything in the reactive world is accomplished with the help of Reactive Streams. In 2013, Netflix, Pivotal, and Lightbend (previously known as Typesafe) felt a need for a standards specification for Reactive Streams as the reactive programming was beginning to spread and more frameworks for reactive programming were starting to emerge, so they started the initiative that resulted in Reactive Streams standard specification, which is now getting implemented across various frameworks and platforms.

You can take a look at the Reactive Streams standard specification at—http://www.reactive-streams.org/.

To write Reactive programs, we need a library or a specific programming language; we can't refer to Kotlin as a reactive language (basically, I don't know any such language that is reactive by itself) as it is a powerful and flexible programming language for modern multiplatform applications, fully interoperable with Java and Android. However, there are reactive libraries out there to help us with these. So, let's take a look at the available list:

You can download and build RxKotlin from GitHub (https://github.com/ReactiveX/RxKotlin). I do not require any other dependencies. The documentation on the GitHub wiki page is well structured. Here's how you can check out the project from GitHub and run the build:

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

You can also use Maven and Gradle, as instructed on the page.

For Gradle, use the following compile dependency:

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

For Maven, use this dependency:

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

This book targets RxKotlin 2.x, so remember to use io.reactive.rxjava2 instead of io.reactivex.rxkotlin, as the latter one is for RxKotlin 1.x.

Now, let's take a look at what RxKotlin is all about. We will begin with something well-known and, gradually, we will get into the secrets of the library.

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:

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.

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.