RxJava: Reactive Extensions for Scala

RxJava: Reactive Extensions For Scala Ben Christensen & Matt Jacobs
Software Engineer – Edge Services at Netﬂix @benjchristensen & @mattrjacobs http://techblog.netﬂix.com/ SF Scala Meetup - October 2013

function composablefunctions reactive reactivelyapplied This presentation is about how the
Netﬂix API application applies a functional programming style in an imperative Java application to apply functions reactively to asynchronously retrieved data ...

function composablefunctions reactive reactivelyapplied ... and transform ...

function composablefunctions reactive reactivelyapplied ... combine and output web service
responses.

function functional lambdas closures (mostly) pure composable asynchronous push events
values reactive We have been calling this approach “functional reactive” since we use functions (lambdas/closures) in a reactive (asynchronous/push) manner.

Observable.from("one", "two", "three") .take(2)
.subscribe((arg) -‐> { System.out.println(arg); }); Java8 Observable("one", "two", "three") .take(2) .subscribe((arg: String) => { println(arg) }) Scala (-‐> (Observable/from ["one" "two" "three"]) (.take 2) (.subscribe (rx/action [arg] (println arg)))) Clojure Observable.from("one", "two", "three") .take(2) .subscribe({arg -‐> println(arg)}) Groovy Observable.from("one", "two", "three") .take(2) .subscribe(lambda { |arg| puts arg }) JRuby Simple examples showing RxJava code in various languages supported by RxJava (https://github.com/Netﬂix/RxJava/tree/master/language-adaptors). Java8 works with rxjava-core and does not need a language-adaptor. It also works with Java 6/7 but without lambdas/closures the code is more verbose.

Observable.from("one", "two", "three") .take(2)
.subscribe((arg) -‐> { System.out.println(arg); }); Java8 Observable("one", "two", "three") .take(2) .subscribe((arg: String) => { println(arg) }) Scala (-‐> (Observable/from ["one" "two" "three"]) (.take 2) (.subscribe (rx/action [arg] (println arg)))) Clojure Observable.from("one", "two", "three") .take(2) .subscribe({arg -‐> println(arg)}) Groovy Observable.from("one", "two", "three") .take(2) .subscribe(lambda { |arg| puts arg }) JRuby Most examples in the rest of this presentation will be in Scala ...

“a library for composing asynchronous and event-based programs using observable
sequences for the Java VM” A Java port of Rx (Reactive Extensions) https://rx.codeplex.com (.Net and Javascript by Microsoft) RxJava http://github.com/Netﬂix/RxJava RxJava is a port of Microsoft’s Rx (Reactive Extensions) to Java that attempts to be polyglot by targeting the JVM rather than just Java the language.

Netﬂix is a subscription service for movies and TV shows
for $7.99USD/month (about the same converted price in each countries local currency).

More than 40 million Subscribers in 50+ Countries and Territories
Netﬂix has over 40 million video streaming customers in 50+ countries and territories across North & South America, United Kingdom, Ireland, Netherlands and the Nordics.

API traffic has grown from ~20 million/day in 2010 to
>2 billion/day 0 500 1000 1500 2000 2010 2011 2012 Today millions of API requests per day

Netflix API Dependency A Dependency D Dependency G Dependency J
Dependency M Dependency P Dependency B Dependency E Dependency H Dependency K Dependency N Dependency Q Dependency C Dependency F Dependency I Dependency L Dependency O Dependency R The Netflix API serves all streaming devices and acts as the broker between backend Netflix systems and the user interfaces running on the 1000+ devices that support Netflix streaming. This presentation is going to focus on why the Netflix API team chose the functional reactive programming model (Rx in particular), how it is used and what benefits have been achieved. Other aspects of the Netflix API architecture can be found at http://techblog.netflix.com/search/label/api and https://speakerdeck.com/benjchristensen/.

Discovery of Rx began with a re-architecture ... More information
about the re-architecture can be found at http://techblog.netﬂix.com/2013/01/optimizing-netﬂix-api.html

... that collapsed network trafﬁc into coarse API calls ...
nested, conditional, concurrent execution Within a single request we now must achieve at least the same level of concurrency as previously achieved by the parallel network requests and preferably better as we can leverage the power of server hardware, lower latency network communication and eliminate redundant calls performed per incoming request.

... and we wanted to allow anybody to create endpoints,
not just the “API Team” User interface client teams now build and deploy their own webservice endpoints on top of the API Platform instead of the “API Team” being the only ones who create endpoints.

We wanted to retain flexibility to use whatever JVM language
we wanted as well as cater to the differing skills and backgrounds of engineers on different teams. Groovy was the first alternate language we deployed in production on top of Java.

Concurrency without each engineer reading and re-reading this → (awesome
book ... everybody isn’t going to - or should have to - read it though, that’s the point)

Owner of api should retain control of concurrency behavior.

What if the implementation needs to change from synchronous to
asynchronous? How should the client execute that method without blocking? spawn a thread? public Data getData(); Owner of api should retain control of concurrency behavior.

public void getData(Callback<T> c); public Future<T> getData(); public Future<List<Future<T>>> getData();
other options ... ? public Data getData();

Iterable pull Observable push T next() throws Exception returns; onNext(T)
onError(Exception) onCompleted() Observable/Observer is the asynchronous dual to the synchronous Iterable/Iterator. More information about the duality of Iterable and Observable can be found at http://csl.stanford.edu/~christos/pldi2010.ﬁt/meijer.duality.pdf and http://codebetter.com/matthewpodwysocki/2009/11/03/introduction-to-the-reactive- framework-part-ii/

onError(Exception) onCompleted() // Iterable[String] // that contains 75 Strings getDataFromLocalMemory() .drop(10) .take(5) .map(_ + "transformed") .foreach(s => println("next => " + s)) // Observable[String] // that emits 75 Strings getDataFromNetwork() .drop(10) .take(5) .map(_ + "transformed") .subscribe(s => println("next => " + s)) The same way higher-order functions can be applied to an Iterable they can be applied to an Observable.

onError(Exception) onCompleted() // Iterable[String] // that contains 75 Strings getDataFromLocalMemory() .drop(10) .take(5) .map(_ + "transformed") .foreach(s => println("next => " + s)) // Observable[String] // that emits 75 Strings getDataFromNetwork() .drop(10) .take(5) .map(_ + "transformed") .subscribe(s => println("next => " + s))

Single Multiple Sync getData: T getData: Iterable[T] Async getData: Future[T]
getData: Observable[T] Grid of synchronous and asynchronous duals for single and multi-valued responses. The Rx Observable is the dual of the synchronous Iterable.

getData: Observable[T] getData match { case "foo" => //do something case "bar" => //do something else case _ => //default } Synchronous scalar response with subsequent conditional logic.

getData: Observable[T] getData.foreach { case "foo" => //do something case "bar" => //do something else case _ => //default } Similar to scalar value except conditional logic happens within a loop.

getData: Observable[T] getData.map { case "foo" => //do something case "bar" => //do something else case _ => //default } Scala Futures are very good at allowing reactive transformation and composition without blocking (unlike java.util.Future) and …

getData: Observable[T] getData.map { case "foo" => //do something case "bar" => //do something else case _ => //default } ... is very similar to the Rx Observable except that an Rx Observable supports multiple values which means it can handle a single value, a sequence of values or an infinite stream. In Scala the usage of the two is very similar, the difference being the ability to handle multiple or infinite values as a stream.

getData: Observable[T] getData.map { case "foo" => //do something case "bar" => //do something else case _ => //default } We wanted to be asynchronous to abstract away the underlying concurrency decisions and composable Futures or Rx Observables are good solutions.

getData: Observable[T] getData.map { case "foo" => //do something case "bar" => //do something else case _ => //default } One reason we chose the Rx Observable is because it gives us a single abstraction that accommodates our needs for both single and multi-valued responses while giving us the higher-order functions to compose nested, conditional logic in a reactive manner that works in a polyglot environment (Java 6+, Groovy, Scala, Clojure, etc)

trait VideoService { def getPersonalizedListOfMovies(userId: Long): VideoList
def getBookmark(userId: Long, videoId: Long): VideoBookmark def getRating(userId: Long, videoId: Long): VideoRating def getMetadata(videoId: Long): VideoMetadata } trait VideoService { def getPersonalizedListOfMovies(userId: Long): Observable[VideoList] def getBookmark(userId: Long, videoId: Long): Observable[VideoBookmark] def getRating(userId: Long, videoId: Long): Observable[VideoRating] def getMetadata(videoId: Long): Observable[VideoMetadata] } ... create an observable api: instead of a blocking api ... With Rx blocking APIs could be converted into Observable APIs and accomplish our architecture goals including abstracting away the control and implementation of concurrency and asynchronous execution.

One of the other positives of Rx Observable was that
it is abstracted from the source of concurrency. It is not opinionated and allows the implementation to decide. For example, an Observable API could just use the calling thread to synchronously execute and respond.

Or it could use a thread-pool to do the work
asynchronously and callback with that thread.

Or it could use multiple threads, each thread calling back
via onNext(T) when the value is ready.

Or it could use an actor pattern instead of a
thread-pool.

Or NIO with an event-loop.

Or a thread-pool/actor that does the work but then performs
the callback via an event-loop so the thread-pool/actor is tuned for IO and event-loop for CPU. All of these diﬀerent implementation choices are possible without changing the signature of the method and without the calling code changing their behavior or how they interact with or compose responses.

client code treats all interactions with the api as asynchronous
the api implementation chooses whether something is blocking or non-blocking and what resources it uses

Observable((observer: Observer[Video]) => { try
{ observer.onNext(Video(id)) observer.onCompleted() } catch { case ex: Throwable => observer.onError(ex) } Subscriptions.empty }) Observable<T> create(Func1<Observer<T>, Subscription> func) Let’s look at how to create an Observable and what its contract is. An Observable receives an Observer and calls onNext 1 or more times and terminates by either calling onError or onCompleted once. More information is available at https://github.com/Netﬂix/RxJava/wiki/Observable

{ observer.onNext(Video(id)) observer.onCompleted() } catch { case ex: Throwable => observer.onError(ex) } Subscriptions.empty }) Observable Observable<T> create(Func1<Observer<T>, Subscription> func) An Observable is created by passing a Func1 implementation...

{ observer.onNext(Video(id)) observer.onCompleted() } catch { case ex: Throwable => observer.onError(ex) } Subscriptions.empty }) Observable<T> create(Func1<Observer<T>, Subscription> func) Observer ... that accepts an Observer ...

{ observer.onNext(Video(id)) observer.onCompleted() } catch { case ex: Throwable => observer.onError(ex) } Subscriptions.empty }) Observable<T> create(Func1<Observer<T>, Subscription> func) ... and when executed (subscribed to) it emits data via ‘onNext’ ...

{ observer.onNext(Video(id)) observer.onCompleted() } catch { case ex: Throwable => observer.onError(ex) } Subscriptions.empty }) Observable<T> create(Func1<Observer<T>, Subscription> func) ... and marks its terminal state by calling ‘onCompleted’ ...

{ observer.onNext(Video(id)) observer.onCompleted() } catch { case ex: Throwable => observer.onError(ex) } Subscriptions.empty }) Observable<T> create(Func1<Observer<T>, Subscription> func) ... or ‘onError’ if a failure occurs. Either ‘onCompleted’ or ‘onError’ must be called to terminate an Observable and nothing can be called after the terminal state occurs. An inﬁnite stream that never has a failure would never call either of these.

def getRating(userId: Long, videoId: Long): Observable[VideoRating] =
Observable(observer => { executor.execute(new Runnable { override def run() { try { val rating = getRatingFromNetwork(userId, videoId) observer.onNext(rating) observer.onCompleted() } catch { case ex: Throwable => observer.onError(ex) } } }) Subscriptions.empty }) Asynchronous Observable with Single Value Example Observable implementation that executes asynchronously on a thread-pool and emits a single value. This explicitly shows an ‘executor’ being used to run this on a separate thread to illustrate how it is up to the Observable implementation to do as it wishes, but Rx always has Schedulers for typical scenarios of scheduling an Observable in a thread-pool or whatever a Scheduler implementation dictates.

def getVideos(): Observable[Video] = Observable(observer => {
try { videos.foreach(observer.onNext) observer.onCompleted() } catch { case ex: Throwable => observer.onError(ex) } Subscriptions.empty }) Synchronous Observable with Multiple Values Caution: This example is eager and will always emit all values regardless of subsequent operators such as take(10) Example Observable implementation that executes synchronously and emits multiple values. Note that the for-loop as implemented here will always complete so should not have any IO in it and be of limited length otherwise it should be done with a lazy iterator implementation or performed asynchronously so it can be unsubscribed from. In Scala it would be preferable to use async/await and in Clojure core.async instead of a blocking loop … or of course it could be done on a separate thread as the previous example.

def getVideos(): Observable[Video] = Observable(observer =>
{ executor.execute(new Runnable { override def run() { try { videoIds.foreach(videoId => { val v = getVideoFromNetwork(videoId) observer.onNext(v) }) observer.onCompleted() } catch { case ex: Throwable => observer.onError(ex) } } }) Subscriptions.empty }) Asynchronous Observable with Multiple Values Example Observable implementation that executes asynchronously on a thread-pool and emits multiple values. Note that for brevity this code does not handle the subscription so will not unsubscribe even if asked. See the ‘getListOfLists' method in the following for an implementation with unsubscribe handled: https://github.com/Netﬂix/RxJava/blob/master/language-adaptors/rxjava-groovy/src/examples/groovy/rx/lang/groovy/examples/ VideoExample.groovy#L125

Asynchronous Observer val subscription = getVideos.subscribe(new rx.Observer[Video] {
def onNext(v: Video) = println("Video: " + v.videoId) def onError(ex: Throwable) { println("Error") ex.printStackTrace } def onCompleted() { println("Completed") } }) Moving to the subscriber side of the relationship we see how an Observer looks. This implements the full interface for clarity of what the types and members are ...

Asynchronous Observer val subscription = getVideos.subscribe(
(v: Video) => println("Video: " + v.videoId), (ex: Throwable) => { println("Error") ex.printStackTrace }, () => println("Completed")) ... but generally the on* method implementations are passed in as functions/lambdas/closures similar to this.

Asynchronous Observer val subscription = getVideos.subscribe(
(v: Video) => println("Video: " + v.videoId), (ex: Throwable) => { println("Error") ex.printStackTrace }) Often the ‘onCompleted’ function is not needed.

functions composable The real power though is when we start
composing Observables together.

Transform: map, flatmap, reduce, scan ... Filter: take, skip, sample,
takewhile, filter ... Combine: concat, merge, zip, combinelatest, multicast, publish, cache, refcount ... Concurrency: observeon, subscribeon Error Handling: onerrorreturn, onerrorresume ... functions composable This is a list of some of the higher-order functions that Rx supports. More can be found in the documentation (https://github.com/Netﬂix/RxJava/wiki) and many more from the original Rx.Net implementation have not yet been implemented in RxJava (but are all listed on the RxJava Github issues page tracking the progress). We will look at some of the important ones for combining and transforming data as well as handling errors asynchronously.

Combining via Merge The ‘merge’ operator is used to combine
multiple Observable sequences of the same type into a single Observable sequence with all data. The X represents an onError call that would terminate the sequence so once it occurs the merged Observable also ends. The ‘mergeDelayError’ operator allows delaying the error until after all other values are successfully merged.

val a: Observable[SomeData] val b: Observable[SomeData] a.merge(b).subscribe(
(element: SomeData) => println("data: " + element))

(element: SomeData) => println("data: " + element)) Each of these Observables are of the same type...

(element: SomeData) => println("data: " + element)) ... and can be represented by these timelines ...

(element: SomeData) => println("data: " + element)) ... that we pass through the ‘merge’ operator ...

(element: SomeData) => println("data: " + element)) ... which looks like this in code ...

(element: SomeData) => println("data: " + element)) ... and emits a single Observable containing all of the onNext events plus the ﬁrst terminal event (onError/onCompleted) from the source Observables ...

(element: SomeData) => println("data: " + element)) ... and these are then subscribed to as a single Observable.

Combining via Zip The ‘zip’ operator is used to combine
Observable sequences of diﬀerent types.

val a: Observable[SomeData] val b: Observable[String] a.zip(b).subscribe(
pair => println("a: " + pair._1 + " b: " + pair._2))

pair => println("a: " + pair._1 + " b: " + pair._2)) Here are 2 Observable sequences with diﬀerent types ...

pair => println("a: " + pair._1 + " b: " + pair._2)) ... represented by 2 timelines with diﬀerent shapes ...

pair => println("a: " + pair._1 + " b: " + pair._2)) ... that we pass through the zip operator that contains a provided function to apply to each set of values received.

pair => println("a: " + pair._1 + " b: " + pair._2)) The transformation function is passed into the zip operator ...

pair => println("a: " + pair._1 + " b: " + pair._2)) ... and in this case is simply taking x & y and combining them into a tuple or pair and then returning it.

pair => println("a: " + pair._1 + " b: " + pair._2)) The output of the transformation function given to the zip operator is emitted in a single Observable sequence ...

pair => println("a: " + pair._1 + " b: " + pair._2)) ... that gives us our pairs when we subscribe to it.

Error Handling val a: Observable[SomeData] val b: Observable[String]
a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed")) If an error occurs then the ‘onError’ handler passed into the ‘subscribe’ will be invoked...

onNext(T) onError(Exception) onCompleted() Error Handling val a: Observable[SomeData] val
b: Observable[String] a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed"))

pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed")) onNext(T) onError(Exception) onCompleted() Error Handling ... but this is the ﬁnal terminal state of the entire composition so we often want to move our error handling to more speciﬁc places. There are operators for that ...

Error Handling The ‘onErrorResumeNext’ operator allows intercepting an ‘onError’ and
providing a new Observable to continue with.

val a: Observable[SomeData] val b: Observable[String] =
a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed"))

getDataB.onErrorResumeNext(getFallbackForB) a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed")) If we want to handle errors on Observable ‘b’ we can compose it with ‘onErrorResumeNext’ and pass in a function that when invoked returns another Observable that we will resume with if onError is called.

getDataB.onErrorResumeNext(getFallbackForB) a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed")) So ‘b’ represents an Observable sequence ...

getDataB.onErrorResumeNext(getFallbackForB) a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed")) ... that emits 3 values ...

getDataB.onErrorResumeNext(getFallbackForB) a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed")) ... and then fails and calls onError ...

getDataB.onErrorResumeNext(getFallbackForB) a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed")) ... which being routed through ‘onErrorResumeNext’ ...

getDataB.onErrorResumeNext(getFallbackForB) a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed")) ... triggers the invocation of ‘getFallbackForB()’ ...

getDataB.onErrorResumeNext(getFallbackForB) a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed")) ... which provides a new Observable that is subscribed to in place of the original Observable ‘b’ ...

getDataB.onErrorResumeNext(getFallbackForB) a.zip(b).subscribe( pair => println("a: " + pair._1 + " b: " + pair._2), ex => println("error occurred: " + ex.getMessage), () => println("completed")) ... so the returned Observable emits a single sequence of 5 onNext calls and a successful onCompleted without an onError.

The ‘onErrorReturn’ operator is similar ...

... except that it returns a speciﬁc value instead of
an Observable. Various ‘onError*’ operators can be found in the Javadoc: http://netﬂix.github.com/RxJava/javadoc/rx/Observable.html

HTTP Request Use Case ObservableHttp.createRequest( HttpAsyncMethods.createGet("http://www.wikipedia.com"),
client) HTTP requests will be used to demonstrate some simple uses of Observable.

client) .toObservable() // Observable[ObservableHttpResponse] The request is lazy and we turn it into an Observable that when subscribed to will execute the request and callback with the response.

client) .toObservable() // Observable[ObservableHttpResponse] .flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }) Once we have the ObservableHttpResponse we can choose what to do with it, including fetching the content which returns an Observable<byte[]>.

ObservableHttp.createRequest( HttpAsyncMethods.createGet("http://www.wikipedia.com"), client)
.toObservable() // Observable[ObservableHttpResponse] .flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }) HTTP Request Use Case We use ﬂatMap as we want to perform nested logic that returns another Observable, ultimately an Observable<String> in this example.

.toObservable() // Observable[ObservableHttpResponse] .flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }) HTTP Request Use Case We use map to transform from byte[] to String and return that.

client) .toObservable() // Observable[ObservableHttpResponse] .flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }).subscribe( (s: String) => System.out.println(s)) We can subscribe to this asynchronously ...

client) .toObservable() // Observable[ObservableHttpResponse] .flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }).subscribe( (s: String) => System.out.println(s)) ... which will execute all of the lazily deﬁned code above and receive String results.

.toObservable() // Observable[ObservableHttpResponse] .flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }) .toBlockingObservable .foreach((s: String) => System.out.println(s)) HTTP Request Use Case Or if we need to be blocking (useful for unit tests or simple demo apps) we can use toBlockingObservable().forEach() to iterate the responses in a blocking manner.

HTTP Request Use Case ObservableHttp.createGet("http://www.wikipedia.com"), client) .toObservable()
// Observable[ObservableHttpResponse] .flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }) This example has shown just a simple request/response.

ObservableHttp.createGet("http://www.wikipedia.com"), client) .toObservable() // Observable[ObservableHttpResponse]
.flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }) HTTP Request Use Case If we change the request ...

ObservableHttp.createGet("http://hostname/hystrix.stream"), client) .toObservable() // Observable[ObservableHttpResponse]
.flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }) HTTP Request Use Case ... to something that streams results (mime-type text/event-stream) we can see a more interesting use of Observable.

ObservableHttp.createGet("http://hostname/hystrix.stream"), client) .toObservable() // Observable[ObservableHttpResponse]
.flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }) .filter((s: String) => s.startsWith(": ping")) .take(30) HTTP Request Use Case We will receive a stream (potentially inﬁnite) of events.

HTTP Request Use Case ObservableHttp.createGet("http://hostname/hystrix.stream"), client) .toObservable()
// Observable[ObservableHttpResponse] .flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }) .filter((s: String) => s.startsWith(": ping")) .take(30) We can ﬁlter out all “: ping” events ...

HTTP Request Use Case ObservableHttp.createGet("http://hostname/hystrix.stream"), client) .toObservable()
// Observable[ObservableHttpResponse] .flatMap((resp: ObservableHttpResponse) => { // access to HTTP status, headers, etc // response.getContent() -‐> Observable[byte[]] val s: rx.Observable[String] = resp.getContent.map((bb: Array[Byte]) => new String(bb)) // Observable[String] s }) .filter((s: String) => s.startsWith(": ping")) .take(30) ... and take the ﬁrst 30 and then unsubscribe. Or we can use operations like window/buﬀer/groupBy/scan to group and analyze the events.

Netflix API Use Case Now we’ll move to a more
involved example of how Rx is used in the Netﬂix API that demonstrates some of the power of Rx to handle nested asynchronous composition.

This marble diagram represents what the code in subsequent slides
is doing when retrieving data and composing the functions.

Observable[Video] emits n videos to onNext() First we start with
a request to fetch videos asynchronously ...

def getVideos(userId: Long): Observable[Map[String, Any]] =
videoService.getVideos(userId) Observable[Video] emits n videos to onNext()

Takes ﬁrst 10 then unsubscribes from origin. Returns Observable [Video]
that emits 10 Videos. def getVideos(userId: Long): Observable[Map[String, Any]] = videoService.getVideos(userId) .take(10) // we only want the first 10 of each list

that emits 10 Videos. The take operator subscribes to the Observable from VideoService.getVideos, accepts 10 onNext calls ...

that emits 10 Videos. ... and then unsubscribes from the parent Observable so only 10 Video objects are emitted from the ‘take’ Observable. The parent Observable receives the unsubscribe call and can stop further processing, or if it incorrectly ignores the unsubscribe the ‘take’ operator will ignore any further data it receives.

The ‘map’ operator allows transforming the input value into a
different output. def getVideos(userId: Long): Observable[Map[String, Any]] = videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .map(video => { // transform video object }) We now apply the ‘map’ operator to each of the 10 Video objects we will receive so we can transform from Video to something else.

Observable<R> b = Observable<T>.map({ T t -‐>
R r = ... transform t ... return r; }) The ‘map’ operator allows transforming from type T to type R.

videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .map(video => { // transform video object }) The ‘map’ operator allows transforming the input value into a different output.

videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .flatMap(video => { // for each video we want to fetch metadata val metadata = video.getMetadata.map(md => // transform to the data and format we want Map("title" -‐> md.get("title"), "length" -‐> md.get("duration"))) // and its rating and bookmark val bookmark = videoService.getBookmark(video, userId).map(b => Map("position" -‐> b.get("position"))) val rating = videoService.getRating(video, userId).map(r => Map("rating" -‐> r.get("rating"))) Observable.zip(Observable(List(metadata, bookmark, rating): _*)).map { case m :: b :: r :: Nil => Map("id" -‐> video.id) ++ m ++ b ++ r } }) We change to ‘flatMap’ which is like merge(map()) since we will return an Observable[T] instead of T. But since we want to do nested asynchronous calls that will result in another Observable being returned we will use flatMap (also knows as mapMany or selectMany) which will flatten an Observable<Observable<T>> into Observable<T> as shown in the following marble diagram ...

Observable<R> b = Observable<T>.mapMany({ T t -‐>
Observable<R> r = ... transform t ... return r; }) flatMap The ‘flatMap’/‘mapMany’ operator allows transforming from type T to type Observable<R>. If ‘map’ were being used this would result in an Observable<Observable<R>> which is rarely what is wanted, so ‘flatMap’/‘mapMany’ flattens this via ‘merge’ back into Observable<R>. This is generally used instead of ‘map’ anytime nested work is being done that involves fetching and returning other Observables.

Observable<R> b = Observable<T>.mapMany({ T t -‐>
Observable<R> r = ... transform t ... return r; }) flatMap A single ﬂattened Observable<R> is returned instead of Observable<Observable<R>>

videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .flatMap(video => { // for each video we want to fetch metadata val metadata = video.getMetadata.map(md => // transform to the data and format we want Map("title" -‐> md.get("title"), "length" -‐> md.get("duration"))) // and its rating and bookmark val bookmark = videoService.getBookmark(video, userId).map(b => Map("position" -‐> b.get("position"))) val rating = videoService.getRating(video, userId).map(r => Map("rating" -‐> r.get("rating"))) Observable.zip(Observable(List(metadata, bookmark, rating): _*)).map { case m :: b :: r :: Nil => Map("id" -‐> video.id) ++ m ++ b ++ r } }) Nested asynchronous calls that return more Observables. Within the ﬂatMap “transformation” function we perform nested asynchronous calls that return more Observables.

videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .flatMap(video => { // for each video we want to fetch metadata val metadata = video.getMetadata.map(md => // transform to the data and format we want Map("title" -‐> md.get("title"), "length" -‐> md.get("duration"))) // and its rating and bookmark val bookmark = videoService.getBookmark(video, userId).map(b => Map("position" -‐> b.get("position"))) val rating = videoService.getRating(video, userId).map(r => Map("rating" -‐> r.get("rating"))) Observable.zip(Observable(List(metadata, bookmark, rating): _*)).map { case m :: b :: r :: Nil => Map("id" -‐> video.id) ++ m ++ b ++ r } }) Nested asynchronous calls that return more Observables. This call returns an Observable<VideoMetadata>.

videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .flatMap(video => { // for each video we want to fetch metadata val metadata = video.getMetadata.map(md => // transform to the data and format we want Map("title" -‐> md.get("title"), "length" -‐> md.get("duration"))) // and its rating and bookmark val bookmark = videoService.getBookmark(video, userId).map(b => Map("position" -‐> b.get("position"))) val rating = videoService.getRating(video, userId).map(r => Map("rating" -‐> r.get("rating"))) Observable.zip(Observable(List(metadata, bookmark, rating): _*)).map { case m :: b :: r :: Nil => Map("id" -‐> video.id) ++ m ++ b ++ r } }) Observable[VideoMetadata] Observable[VideoBookmark] Observable[VideoRating] 3 separate types are being fetched asynchronously and each return an Observable.

videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .flatMap(video => { // for each video we want to fetch metadata val metadata = video.getMetadata.map(md => // transform to the data and format we want Map("title" -‐> md.get("title"), "length" -‐> md.get("duration"))) // and its rating and bookmark val bookmark = videoService.getBookmark(video, userId).map(b => Map("position" -‐> b.get("position"))) val rating = videoService.getRating(video, userId).map(r => Map("rating" -‐> r.get("rating"))) Observable.zip(Observable(List(metadata, bookmark, rating): _*)).map { case m :: b :: r :: Nil => Map("id" -‐> video.id) ++ m ++ b ++ r } }) Each Observable transforms its data using ‘map’ Each of the 3 diﬀerent Observables are transformed using ‘map’, in this case from the VideoMetadata type into a dictionary of key/value pairs.

For each of the 10 Video objects it transforms via
‘mapMany’ function that does nested async calls.

For each Video ‘v’ it calls getMetadata() which returns Observable[VideoMetadata]
These nested async requests return Observables that emit 1 value.

The Observable[VideoMetadata] is transformed via a ‘map’ function to return
a Map of key/values.

Same for Observable[VideoBookmark] and Observable[VideoRating] Each of the .map() calls
emits the same type (represented as an orange circle) since we want to combine them later into a single dictionary (Map).

videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .flatMap(video => { // for each video we want to fetch metadata val metadata = video.getMetadata.map(md => // transform to the data and format we want Map("title" -‐> md.get("title"), "length" -‐> md.get("duration"))) // and its rating and bookmark val bookmark = videoService.getBookmark(video, userId).map(b => Map("position" -‐> b.get("position"))) val rating = videoService.getRating(video, userId).map(r => Map("rating" -‐> r.get("rating"))) // compose these together Observable.zip(Observable(List(metadata, bookmark, rating): _*)).map { case m :: b :: r :: Nil => // now transform to complete dictionary // of data we want for each Video Map("id" -‐> video.id) ++ m ++ b ++ r } })

videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .flatMap(video => { // for each video we want to fetch metadata val metadata = video.getMetadata.map(md => // transform to the data and format we want Map("title" -‐> md.get("title"), "length" -‐> md.get("duration"))) // and its rating and bookmark val bookmark = videoService.getBookmark(video, userId).map(b => Map("position" -‐> b.get("position"))) val rating = videoService.getRating(video, userId).map(r => Map("rating" -‐> r.get("rating"))) // compose these together Observable.zip(Observable(List(metadata, bookmark, rating): _*)).map { case m :: b :: r :: Nil => // now transform to complete dictionary // of data we want for each Video Map("id" -‐> video.id) ++ m ++ b ++ r } }) The ‘zip’ operator combines the 3 asynchronous Observables into 1 We use ‘zip’ to combine the 3 together and apply a function to transform them into a single combined format that we want, in this case a dictionary that contains the key values pairs from the dictionaries emitted by ‘metadata’, ‘bookmark’, and ‘ratings’ along with the videoId also available within scope of the ﬂatMap function and ‘closed over’ by the closure being executed in ‘zip’.

Observable.zip(a, b, { a, b, -‐>
... operate on values from both a & b ... return [a, b]; // i.e. return tuple })

videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .flatMap(video => { // for each video we want to fetch metadata val metadata = video.getMetadata.map(md => // transform to the data and format we want Map("title" -‐> md.get("title"), "length" -‐> md.get("duration"))) // and its rating and bookmark val bookmark = videoService.getBookmark(video, userId).map(b => Map("position" -‐> b.get("position"))) val rating = videoService.getRating(video, userId).map(r => Map("rating" -‐> r.get("rating"))) // compose these together Observable.zip(Observable(List(metadata, bookmark, rating): _*)).map { case m :: b :: r :: Nil => // now transform to complete dictionary // of data we want for each Video Map("id" -‐> video.id) ++ m ++ b ++ r } }) return a single Map (dictionary) of transformed and combined data from 4 asynchronous calls

videoService.getVideos(userId) .take(10) // we only want the first 10 of each list .flatMap(video => { // for each video we want to fetch metadata val metadata = video.getMetadata.map(md => // transform to the data and format we want Map("title" -‐> md.get("title"), "length" -‐> md.get("duration"))) // and its rating and bookmark val bookmark = videoService.getBookmark(video, userId).map(b => Map("position" -‐> b.get("position"))) val rating = videoService.getRating(video, userId).map(r => Map("rating" -‐> r.get("rating"))) // compose these together Observable.zip(Observable(List(metadata, bookmark, rating): _*)).map { case m :: b :: r :: Nil => // now transform to complete dictionary // of data we want for each Video Map("id" -‐> video.id) ++ m ++ b ++ r } }) return a single Map (dictionary) of transformed and combined data from 4 asynchronous calls The entire composed Observable emits 10 Maps (dictionaries) of key/value pairs for each of the 10 VIdeo objects it receives.

The ‘mapped’ Observables are combined with a ‘zip+map’ function that
emits a Map (dictionary) with all data. The entire composed Observable emits 10 Maps (dictionaries) of key/value pairs for each of the 10 VIdeo objects it receives.

The full sequence returns Observable[Map] that emits a Map (dictionary)
for each of 10 Videos.

interactions with the api are asynchronous and declarative api implementation
controls concurrency behavior

/ps3/home Dependency F 10 Threads Dependency G 10 Threads Dependency
H 10 Threads Dependency I 5 Threads Dependency J 8 Threads Dependency A 10 Threads Dependency B 8 Threads Dependency C 10 Threads Dependency D 15 Threads Dependency E 5 Threads Dependency K 15 Threads Dependency L 4 Threads Dependency M 5 Threads Dependency N 10 Threads Dependency O 10 Threads Dependency P 10 Threads Dependency Q 8 Threads Dependency R 10 Threads Dependency S 8 Threads Dependency T 10 Threads /android/home /tv/home Functional Reactive Dynamic Endpoints Asynchronous Java API We have found Rx to be a good ﬁt for creating Observable APIs and composing asynchronous data together while building web services using this approach.

/ps3/home Dependency F 10 Threads Dependency G 10 Threads Dependency
H 10 Threads Dependency I 5 Threads Dependency J 8 Threads Dependency A 10 Threads Dependency B 8 Threads Dependency C 10 Threads Dependency D 15 Threads Dependency E 5 Threads Dependency K 15 Threads Dependency L 4 Threads Dependency M 5 Threads Dependency N 10 Threads Dependency O 10 Threads Dependency P 10 Threads Dependency Q 8 Threads Dependency R 10 Threads Dependency S 8 Threads Dependency T 10 Threads /android/home /tv/home Functional Reactive Dynamic Endpoints Asynchronous Java API Hystrix fault-isolation layer With the success of Rx at the top layer of our stack we’re now ﬁnding other areas where we want this programming model applied.

+ val user: Observable[User] = Observable(new GetUserCommand(userId).observe())
val geo: Observable[Geo] = Observable(new GetGeoCommand(request).observe()) user.zip(geo).map { case (u, g) => Map("username" -‐> u.username, "currentLocation" -‐> g.country) } RxJava in Hystrix 1.3+ https://github.com/Netﬂix/Hystrix One example of us pushing Rx deeper into our stack is the addition of support for RxJava to Hystrix version 1.3. More information on the release can be found at https://github.com/Netﬂix/Hystrix/releases/tag/1.3.0

observable apis Looking back, Rx has enabled us to achieve
our goals that started us down this path.

lessons learned Developer Training & Documentation As we implemented and
adopted Rx and enabled dozens of developers (most of them of either Javascript or imperative Java backgrounds) we found that workshops, training sessions and well-written documentation was very helpful in “onboarding” them to the new approach. We have found it generally takes a few weeks to get adjusted to the style.

Developer Training & Documentation Debugging and Tracing lessons learned Asynchronous
code is challenging to debug. Improving our ability to debug, trace and visualize Rx “call graphs” is an area we are exploring.

Developer Training & Documentation Debugging and Tracing Only “rule” has
been “don’t mutate state outside of function” lessons learned Generally the model has been self-governing (get the code working and all is ﬁne) but there has been one principle to teach since we are using this approach in mutable, imperative languages - don’t mutate state outside the lambda/ closure/function.

functional lambdas closures (mostly) pure composable asynchronous push events values
reactive The Rx “functional reactive” approach is a powerful and straight-forward abstraction for asynchronously composing values and events and has worked well for the Netﬂix API.

Functional Reactive in the Netflix API with RxJava http://techblog.netflix.com/2013/02/rxjava-netflix-api.html Optimizing
the Netflix API http://techblog.netflix.com/2013/01/optimizing-netflix-api.html RxJava & Scala Adaptor https://github.com/Netflix/RxJava https://github.com/Netflix/RxJava/tree/master/language-adaptors/rxjava-scala @RxJava Ben Christensen @benjchristensen http://www.linkedin.com/in/benjchristensen jobs.netflix.com

RxJava: Reactive Extensions for Scala

RxJava: Reactive Extensions for Scala

More Decks by Ben Christensen

Other Decks in Programming

Featured

Transcript