Functional Reactive Programming / Compositional Event Systems

Taming Asynchronous Workﬂows with Functional Reactive Programming EuroClojure - Kraków, 2014 Leonardo Borges @leonardo_borges www.leonardoborges.com www.thoughtworks.com

About ‣ ThoughtWorker ‣ Functional Programming & Clojure advocate ‣ Founder of the Sydney Clojure User Group ‣ Currently writing “Clojure Reactive Programming”

Taming Asynchronous Workﬂows with Functional Reactive Programming

Taming Asynchronous Workﬂows with Functional Reactive Programming Compositional Event Systems

http://bit.ly/reactive-cocoa-commit

There are only two hard things in Computer Science: cache invalidation and naming things. - Phil Karlton :)

Ok, so what’s the difference?

‣ Created in 1997 by Conal Elliott for the reactive animations framework Fran, in Haskell ‣ Since then other implementations have appeared: reactive-banana, NetWire, Sodium (all in Haskell) ‣ And then FRP-inspired ones: Rx[.NET | Java | JS], Baconjs, reagi (Clojurescript) ‣ Main abstractions: Behaviors e Events More about FRP

‣ Created in 1997 by Conal Elliott for the reactive animations framework Fran, in Haskell ‣ Since then other implementations have appeared: reactive-banana, NetWire, Sodium (all in Haskell) ‣ And then FRP-inspired ones: Rx[.NET | Java | JS], Baconjs, reagi (Clojure[script]) ‣ Main abstractions: Behaviors e Events ‣ Traditionally deﬁned as: type Behavior a = [Time] -> [a]! type Event a = [Time] -> [Maybe a] More about FRP

We’ll be focusing on Compositional Event Systems

Imperative programming describes computations as a series of actions which modify program state var result = 1;! numbers.forEach(function(n){! if(n % 2 === 0) {! result *= n;! }! });! console.log( result );! // 8! var numbers = [1,2,3,4,5]; Requires a variable to store state

var result = 1;! numbers.forEach(function(n){! if(n % 2 === 0) {! result *= n;! }! });! console.log( result );! // 8! var numbers = [1,2,3,4,5]; We iterate over the array Imperative programming describes computations as a series of actions which modify program state

var result = 1;! numbers.forEach(function(n){! if(n % 2 === 0) {! result *= n;! }! });! console.log( result );! // 8! var numbers = [1,2,3,4,5]; And then we ﬁlter the items… Imperative programming describes computations as a series of actions which modify program state

var result = 1;! numbers.forEach(function(n){! if(n % 2 === 0) {! result *= n;! }! });! console.log( result );! // 8! var numbers = [1,2,3,4,5]; …and perform the multiplication in the same function Imperative programming describes computations as a series of actions which modify program state

(def numbers [1 2 3 4 5])! ! (def result! (->> numbers! (filter even?)! (reduce *)))! ! (prn result)! ! ;; 8 In functional programming, we describe what we want to do but not how we want it done

That is, there are no variables with local state and we get better re-use from single purpose functions

Compositional Event Systems brings the same principle to values we work with daily: DOM events (clicks, key presses, mouse movement), Ajax calls…

Game movements in Javascript var JUMP = 38, CROUCH = 40,! LEFT = 37, RIGHT = 39,! FIRE = 32;! ! function goRight (){! $(‘h1').html("Going right...");! }! ! function goLeft (){! $(‘h1').html("Going left...");! }! ! function jump (){! $('h1').html("Jumping...");! }! ! function crouch (){! $('h1').html("Crouching...");! }! ! function fire (){! $('h1').html("Firing...");! }

Game movements in Javascript $(window.document).keyup(function(event){! switch(event.keyCode){! case JUMP :! jump();! break;! case CROUCH:! crouch();! break;! case LEFT :! goLeft();! break;! case RIGHT :! goRight();! break;! case FIRE :! fire();! break;! };! });

We now have limitations similar to the multiplication example

Let us think key presses as a list of keys over a period of time

This leads to the following solution

Reactive game movements (def UP 38) (def RIGHT 39)! (def DOWN 40) (def LEFT 37)! (def FIRE 32) (def PAUSE 80)! ! ! (def source (.fromEvent js/Rx.Observable js/window "keyup"));! ! (-> source (.filter #(= UP %)) (.subscribe jump))! (-> source (.filter #(= DOWN %)) (.subscribe crouch))! (-> source (.filter #(= RIGHT %)) (.subscribe go-right))! (-> source (.filter #(= LEFT %)) (.subscribe go-left))! (-> source (.filter #(= FIRE %)) (.subscribe fire))! http://bit.ly/rxjava-github http://bit.ly/rxjs-github

Behaviours ;; behavior examples! (def time-b (r/behavior (System/currentTimeMillis)))! @time-b! ;; 1403691712988! @time-b! ;; 1403691714156! http://bit.ly/reagi

Behaviours to Event Streams http://bit.ly/reagi (def time-e (r/sample 1000 time-b))! ! (->> time-e (r/map println))! ;; t + 1 sec! ;; 1403692132586! ;; t + 2 sec:! ;; 1403692133587!

Combinators (-> (Observable/return 42)! (.map #(* % 2))! (.subscribe println))! ! ;; 84! ! (-> (Observable/from [10 20 30])! (.map #(* % 2))! (.reduce +)! (.subscribe println))! ! ;; 120!

Combinators: ﬂatMap / selectMany (defn project-range [n]! (Rx.Observable/return (range n)))! ! (-> (Observable/from [1 2 3])! (.selectMany project-range)! (.subscribe (rx/fn* println)))! ! ;; 0! ;; 0! ;; 1! ;; 0! ;; 1! ;; 2!

(Observable/from [1 2 3]) 1 2 3

(-> (Observable/from [1 2 3])! (.selectMany project-range)! …) (project-range 2) 0 1 (project-range 1) 0 (project-range 3) 0 1 2 0 0 1 0 1 2

What about network IO? ‣ Callback hell :( ‣ Clojure promises don’t compose ‣ Promises in JS are slightly better but have limitations ‣ They work well for a single level of values ‣ However are a poor composition mechanism ‣ What if we have a series of values that changes over time?

This is what the server gives us {:id 7! :question "Which is the best music style?"! :results {:a 10! :b 47! :c 17}}!

And this is what we want ‣ Render results ‣ Continuously poll server every 2 secs ‣ If current question is the same as the previous one update results; Otherwise: ‣ Stop polling; ‣ Display countdown message; ‣ Render new question and results; ‣ Restart polling;

First, we need to turn the results into a stream 4 3 3 2 1 1

So we duplicate the stream, skipping one element 4 3 3 2 1 1 5 4 3 3 2 1 (skip 1)

Finally, we zip the streams 4 3 3 2 1 1 5 4 3 3 2 1 zip [5,4] [4,3] [3,3] [3,2] [2,1] [1,1]

The core idea, in code (defn results-observable! "Returns an Observable that yields server-side questions/results"! []! (.create js/Rx.Observable! (fn [observer]! (srm/rpc! (poll-results) [resp]! (.onNext observer resp))! (fn [] (.log js/console "Disposed")))))

The core idea, in code (def results-connectable! "Zips results-observable with itself, but shifted by 1.! This simulates a 'buffer' or 'window' of results"! (let [obs (-> js/Rx.Observable! (.interval 2000)! (.selectMany results-observable)! (.publish)! (.refCount))! obs-1 (.skip obs 1)]! (.zip obs obs-1 (fn [prev curr]! {:prev prev! :curr curr}))))! Turn results into a stream

The core idea, in code (def results-connectable! "Zips results-observable with itself, but shifted by 1.! This simulates a 'buffer' or 'window' of results"! (let [obs (-> js/Rx.Observable! (.interval 2000)! (.selectMany results-observable)! (.publish)! (.refCount))! obs-1 (.skip obs 1)]! (.zip obs obs-1 (fn [prev curr]! {:prev prev! :curr curr}))))! Clone stream, skip one

The core idea, in code (def results-connectable! "Zips results-observable with itself, but shifted by 1.! This simulates a 'buffer' or 'window' of results"! (let [obs (-> js/Rx.Observable! (.interval 2000)! (.selectMany results-observable))! obs-1 (.skip obs 1)]! (.zip obs obs-1 (fn [prev curr]! {:prev prev! :curr curr}))))! Zip them together

Buffering (def results-buffer! "Returns an Observable with results buffered into a 2-element vector"! (-> js/Rx.Observable! (.interval 2000)! (.selectMany results-observable)! (.bufferWithCount 2)))!

"FRP is about handling time-varying values like they were regular values" - Haskell Wiki

"FRP is about handling time-varying values like they were regular values" - Haskell Wiki It also applies to FRP-inspired systems

Previously, with CES (-> (Observable/from [10 20 30])! (.map (rx/fn [v] (* v 2)))! (.reduce (rx/fn* +)! (.subscribe (rx/fn* println)))))

With core.async (defn from-array [coll]! (let [stream-c (chan)]! (go (doseq [n coll]! (>! stream-c n))! (close! stream-c))! stream-c))! ! (def c (->> (async-from [10 20 30])! (a/map< #(* % 2))! (a/reduce + 0)))! ! (go-loop []! (when-let [v (<! c)]! (println v)! (recur)))!

Multiple subscribers with CES (def sum-of-squares (-> (Observable/from [10 20 30])! (.map (rx/fn [v] (* v 2)))! (.reduce (rx/fn* +))))! ! ! ! (.subscribe sum-of-squares (rx/action* println)) ;; 120! (.subscribe sum-of-squares (rx/action* println)) ;; 120

Multiple subscribers with core.async [1/3] (def in (chan))! (def sum-of-squares (->> in! (a/map< #(* % 2))! (a/reduce + 0)))!

Multiple subscribers with core.async [2/3] (def publication (pub sum-of-squares (fn [_] :n)))! ! (def sub-1 (chan))! (def sub-2 (chan))! ! (sub publication :n sub-1)! (sub publication :n sub-2)

Multiple subscribers with core.async [3/3] (go (doseq [n [10 20 30]]! (>! in n))! (close! in))! ! (go-loop []! (when-let [v (<! sub-1)]! (prn v)! (recur))) ;; 120! ! (go-loop []! (when-let [v (<! sub-2)]! (prn v)! (recur))) ;; 120

core.async operates at a lower level of abstraction

it is however a great foundation for a FRP-inspired framework

Reagi - shown earlier - is built on top of core.async http://bit.ly/reagi

Bonus example: a reactive API to AWS

Bonus example: a reactive API to AWS ‣ Retrieve list of resources from a stack (CloudFormation.describeStackResources) ‣ For each EC2 Instance, call EC2.describeInstances to retrieve status ‣ For each RDS Instance, call RDS.describeDBInstances to retrieve status ‣ Merge results and display

Step 1: turn api calls into streams (defn resources-stream [stack-name]! (.create js/Rx.Observable! (fn [observer]! (.describeStackResources js/cloudFormation #js {"StackName" : stackName}! (fn [err data]! (if err! (.onError observer err)! (doseq [resource data]! (.onNext observer resource))! (.onCompleted observer)))! (fn [] (.log js/console "Disposed")))))

Step 1: turn api calls into streams (defn ec2-instance-stream [resource-ids]! (.create js/Rx.Observable! (fn [observer]! (.describeInstaces js/ec2 #js {"InstanceIds" resource-ids}! (fn [err data]! (if err! (.onError observer err)! (doseq [instance data]! (.onNext observer instance)))! (.onCompleted observer)))! (fn [] (.log js/console "Disposed")))))!

Step 1: turn api calls into streams (defn rds-instance-stream [resource-id]! (.create js/Rx.Observable! (fn [observer]! (.describeDBInstances js/rds #js {"DBInstanceIdentifier" resource-id}! (fn [err data]! (if err! (.onError observer err)! (.onNext observer data))! (.onCompleted observer)))! (fn [] (.log js/console "Disposed")))))

Step 2: transform the different API responses into a common output (def resources (resourcesStream "my-stack"))! ! (def ec2-data (-> resources! (.filter ec2?)! (.map :resource-id)! (.flatMap ec2-instance-stream)! (.map (fn [data] {:instance-id ...! :status ...}))))! ! (def rds-data (-> resources! (.filter rds?)! (.map :resource-id)! (.flatMap rds-instance-stream)! (.map (fn [data] {:instance-id ...! :status ...}))))!

Step 3: merge results and update UI (-> ec2-data! (.merge rds-data)! (.reduce conj [])! (.subscribe (fn [data] (update-interface ...))))

Easy to reason about, maintain and test

References ! ‣ Conal Elliott “Functional Reactive Animation” paper: http://bit.ly/conal-frp ! FRP-inspired frameworks: ‣ Reagi: http://bit.ly/reagi ‣ RxJS: http://bit.ly/rxjs-github ‣ RxJava: http://bit.ly/rxjava-github ‣ Bacon.js: https://github.com/baconjs/bacon.js ! FRP implementations: ‣ Reactive Banana: http://www.haskell.org/haskellwiki/Reactive-banana ‣ Elm: http://elm-lang.org/ ‣ NetWire: http://www.haskell.org/haskellwiki/Netwire

Thanks! Questions? Leonardo Borges @leonardo_borges www.leonardoborges.com www.thoughtworks.com

Functional Reactive Programming / Compositional Event Systems

More Related Content

What's hot

Viewers also liked

Similar to Functional Reactive Programming / Compositional Event Systems

More from Leonardo Borges

Recently uploaded

Functional Reactive Programming / Compositional Event Systems