So what is functional reactive programming?

‘The application of functions¹ in a reactive² manner to asynchronous data sequences’ - RxJava Wiki

[1] lambdas & closures

[2] asynchronous & push

In the beginning

We had single values

getUser: User

Synchronous + Blocking

Easy to use, but doesn't scale to multiple values

Next we got many values in one call

getUsers: List[User]

Synchronous + Blocking

To work with many values we use Iterators

hasNext: Boolean
next: User

Iterators lead to composable data manipulation

val adults = users.filter(_.age > 17).sortBy(p => (p.lName, p.fName)).take(10)

but we block until the full list is ready

Normal Execution

+-----------------------------------------------------------------------------+
| execute typical instruction                   1 nanosec (1 billionth/second)|
| fetch from L1 cache memory                  0.5 nanosec                     |
| branch misprediction                          5 nanosec                     |
| fetch from L2 cache memory                    7 nanosec                     |
| Mutex lock/unlock                            25 nanosec                     |
| fetch from main memory                      100 nanosec                     |
+-----------------------------------------------------------------------------+
| send 2K bytes over 1Gbps network         20,000 nanosec                     |
| read 1MB sequentially from memory       250,000 nanosec                     | 
| fetch from new disk location (seek)   8,000,000 nanosec                     |
| read 1MB sequentially from disk      20,000,000 nanosec                     |
| send packet US to Europe and back   150,000,000 nanosec (150 milliseconds)  |
+-----------------------------------------------------------------------------+

source: http://norvig.com/21-days.html#answers

Blocking Requests

Normal Execution

+-----------------------------------------------------------------------------+
| execute typical instruction                   1 seconds                     |
| fetch from L1 cache memory                  0.5 seconds                     |
| branch misprediction                          5 seconds                     |
| fetch from L2 cache memory                    7 seconds                     |
| Mutex lock/unlock                            30 seconds                     |
| fetch from main memory                      1.5 minutes                     |
+-----------------------------------------------------------------------------+
| send 2K bytes over 1Gbps network            5.5 hours                       |
| read 1MB sequentially from memory             3 days                        | 
| fetch from new disk location (seek)          13 weeks                       |
| read 1MB sequentially from disk             6.5 months                      |
| send packet US to Europe and back             5 years                       |
+-----------------------------------------------------------------------------+

source: https://class.coursera.org/reactive-001/lecture/51

Blocking Requests

Callbacks/CPS let us work asynchronously

And lead to Callback Hell

getTweetsWithUrlsFor("tednaleid", function(err, tweets) {
  if (err) console.log("Error getting tweets for: " + err);
  else {
    var mostRecentShortUrl = parseTweetsForUrls(tweets[0]);
    httpGet(mostRecentShortUrl, function(err, responseBody) {
      if (err) console.log("Error retrieving tweet");
      else {
        var targetUrl = extractUrlFromTweet(responseBody);
        httpGet(targetUrl, function(err, responseBody) {
          if (err) console.log("Error retrieving tweet");
          else console.log("most recent linked text: " + responseBody);
        });
      }
    });
  }
});

Futures/Promises can help

getTweetsWithUrlsFor("tednaleid").then(function(tweets) {
  var mostRecentShortUrl = parseTweetsForUrls(tweets[0]);
  return httpGet(mostRecentShortUrl);
}).then(function(responseBody) {
  var targetUrl = extractUrlFromTweet(responseBody);
  return httpGet(targetUrl);
}).then(
  function(responseBody) {
    console.log("most recent linked text: " + responseBody);
  },
  function(error) {
    console.error("Problem getting most recent text: " + error);
  }
);

Get single Future value

getUser: Future[User]

Asynchronous + Non-Blocking

What if you need a list of items asynchronously?

One way: Future containing an Iterator

getUsers: Future[List[Users]]

Issue: Need to wait for full list

Result: Slow user experience

Issue: What if full list is too big?

Result: Need to paginate

Issue: What if it's a stream that has updates?

Result: We need to poll for updates and merge them in

Another way: Observer pattern

Observer - Swing Event handling

var path: Path = null
val moveObserver = { (event: MouseEvent) =>
    path.lineTo(event.position) 
    draw(path)
}
control.addMouseDownObserver { event =>
    path = new Path(event.position)
    control.addMouseMoveObserver(moveObserver) 
}
control.addMouseUpObserver { event => 
    control.removeMouseMoveObserver(moveObserver)
    path.close()
    draw(path)
}

from: http://lampwww.epfl.ch/~imaier/pub/DeprecatingObserversTR2010.pdf

Multiple Observers needed to make a State Machine

var path: Path = null
val moveObserver = { (event: MouseEvent) =>  // <- Observer #1
    path.lineTo(event.position) 
    draw(path)
}
control.addMouseDownObserver { event =>      // <- Observer #2
    path = new Path(event.position)
    control.addMouseMoveObserver(moveObserver) 
}
control.addMouseUpObserver { event =>        // <- Observer #3
    control.removeMouseMoveObserver(moveObserver)
    path.close()
    draw(path)
}

Observers promote side effects/Global state

var path: Path = null                // <- Global State shared by all Observers
val moveObserver = { (event: MouseEvent) =>
    path.lineTo(event.position)      // mutate state
    draw(path)
}
control.addMouseDownObserver { event =>
    path = new Path(event.position)  // mutate state
    control.addMouseMoveObserver(moveObserver) 
}
control.addMouseUpObserver { event => 
    control.removeMouseMoveObserver(moveObserver)
    path.close()                     // mutate state
    draw(path)
}

Resource Management spread across Observers

var path: Path = null                
val moveObserver = { (event: MouseEvent) =>
    path.lineTo(event.position)     
    draw(path)
}
control.addMouseDownObserver { event =>
    path = new Path(event.position)          // <- Resource Instantiation
    control.addMouseMoveObserver(moveObserver) 
}
control.addMouseUpObserver { event => 
    control.removeMouseMoveObserver(moveObserver)
    path.close()                             // <- Resource Cleanup
    draw(path)
}

Cleanup not guaranteed to be called

Is there a better way?

What if this happened?

You got your Iterator in my Observer!

The result would be the Observable

Two great tastes that taste great together!

Iterator : pull :: Observable : push

                                Iterator (pull)     Observable (push)
                             +-------------------+--------------------+
              retrieve data  | T next()          | onNext(T)          |
                             +-------------------+--------------------+
               error occurs  | throws Exception  | onError(Exception) |
                             +-------------------+--------------------+
          end of collection  | return            | onCompleted()      |
                             +-------------------+--------------------+

(RxJava implementation)

Observables allow you to asynchronously ‘get’

getUsers: Observable[User]

They are also composable, as we will see

How is the Observable returned by ‘getUsers’ implemented?

Using the same thread? A different thread? Multiple threads? An Actor or a Channel?

Implementation details!

The caller doesn't care, it only wants an Observable!

Using Observables

subscribe

val numbers = Observable(1, 2, 3, 4, 5)


numbers.subscribe(
  (x: Int) => println(x),                   // onNext
  (e: Throwable) => println("Error! " + e), // onError
  () => println("Done!")                    // onSuccess
)

prints

1
2
3
4
5
Done!

map

from https://github.com/Netflix/RxJava/wiki/Transforming-Observables#map

map

val numbers = Observable(1, 2, 3, 4, 5)

numbers.map(x => x * 2).subscribe(
  (x: Int) => println(x),                   // onNext
  (e: Throwable) => println("Error! " + e), // onError
  () => println("Done!")                    // onSuccess
)

prints

2
4
6
8
10
Done!

reduce

from https://github.com/Netflix/RxJava/wiki/Transforming-Observables#reduce-or-aggregate

reduce

val numbers = Observable(1, 2, 3, 4, 5)

numbers.reduce((a, b) => a + b).subscribe(
  (x: Int) => println(x),                   // onNext
  (e: Throwable) => println("Error! " + e), // onError
  () => println("Done!")                    // onSuccess
)

prints

15
Done!

filter

from https://github.com/Netflix/RxJava/wiki/Filtering-Observables#filter-or-where

filter

val moreNumbers = Observable(1, 2, 3, 4, 5, 6, 7, 8, 9)

moreNumbers.filter(x => x % 2 == 0).subscribe(
  (x: Int) => println(x),                   // onNext
  (e: Throwable) => println("Error! " + e), // onError
  () => println("Done!")                    // onSuccess
)

prints

2
4
6
8
Done!

merge

from https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge

merge

val odds = Observable(1, 3, 5)
val evens = Observable(2, 4, 6)

odds.merge(evens).subscribe(
  (x: Int) => println(x),                   // onNext
  (e: Throwable) => println("Error! " + e), // onError
  () => println("Done!")                    // onSuccess
)

prints

1
3
5
2
4
6
Done!

(order isn't normally deterministic)

zip

from https://github.com/Netflix/RxJava/wiki/Combining-Observables#zip

zip

val odds = Observable(1, 3, 5)
val evens = Observable(2, 4, 6)

odds.zip(evens).subscribe(
  pair => println(s"odd: ${pair._1}, even: ${pair._2}"), // onNext
  (e: Throwable) => println("Error! " + e),              // onError
  () => println("Done!")                                 // onSuccess
)

prints

odd: 1, even: 2
odd: 3, even: 4
odd: 5, even: 6
Done!

in Scala zip emits a tuple by default

All functions return new Observables and are free from side-effects

Composition of functions

val numbers = Observable(1, 2, 3, 4, 5)

numbers.take(3).flatMap(x => Observable(x, x * 2)).subscribe(println(_))

prints

1
2
2
4
3
6

(flatMap! Observable is a monad)

Real-World Example: Netflix API

def getVideos(userId: Long): Observable[Map[String, Any]] = 
  videoService.getVideos(userId)                             // async Observable
     // take the first 10, then auto-unsubscribe
    .take(10)  
    .flatMap(video => {
      val metadata = video.getMetaData(video)                // async Map
      val bookmark = videoService.getBookmark(video, userId) // async Map
      val rating = videoService.getRating(video, userId)     // async Map
      // zip the data from those 3 services into an Observable
      Observable.zip(Observable(List(metadata, bookmark, rating): _*)).map {
        case m :: b :: r :: Nil => Map("id" -> video.id) ++ m ++ b ++ r
      }
  })

That method composes 4 separate async calls into a List of 10 Maps

Map(id -> 1, rating -> *****, pos -> 1:33, length -> 2h, title -> Gravity)
…

‘Four essential effects in programming’ - Erik Meijer

                                         One           Many       
                                    +------------+---------------+
             Synchronous/Blocking   |     T      |  Iterable[T]  |
                                    +------------+---------------+
         Asynchronous/Non-Blocking  | Future[T]  | Observable[T] |
                                    +------------+---------------+

FRP is a great fit for:

Handling UI Events like mouse actions or keyboard input

Reacting to streams of information

ex: Integration events from a message bus or listening on a WebSocket

FRP can work well for asynchronous tasks

(especially those that aggregate data)

If you're using FRP in place of a regular Iterator

You're doing it wrong

Implementations of Functional Reactive Programming

Rx - Microsoft

CLR & JavaScript

RxJava - Netflix

Java, Scala, Groovy, Clojure, Kotlin, JRuby

Elm

Language that compiles to JavaScript

Another language that compiles to JavaScript

JavaScript Library

Other similar tech that addresses the problems

Scala - Actors (Akka)

Go - Goroutines/Channels

Clojure - core.async

Groovy - Reactor

Demo

Questions?