Reactive Programming

with JavaScript

André Werlang (@awerlang) - Porto Alegre - RSJS April, 2017

Ou:

Pra que serve essa variável perdida aqui mesmo? Abordando o problema reativamente!

What we are about to see

  • Decoupling
  • Composition
  • Asynchronous workflows
  • Vocabulary

Attempts

Callbacks

  • Just pass in a function
  • It can be synchronous...
  • ...but probably not
  • ...which leads to terrible coding style
  • And ties producer & consumer

Promises

  • ES2015
  • Abstraction around a future, immutable value
  • First-class citizen
  • They're eager
  • All or nothing, there's no progress indication
  • Does not provide for cancelling
  • Bad style still leads to callback hell

Generators

  • ES2015
  • Synchronous (or cooperative)...
  • ...but can yield a promise
  • Enforces an imperative-style

Generators


function* getAddresses(action) {
  try {
    const response = yield call(API.fetchAddress, ADDRESSES_ENDPOINT);
    const data = {
      options: [...response.data]
    };
    yield put({type: FINISH_LOCATION_SEARCH, payload: response.data});
  } catch (err) {
    yield put({
      type: LOCATION_SEARCH_FAILURE,
      payload: {
        _error: err.message
      }
    });
  }
}

Async/await

  • ES2017
  • Designed on top of promises
  • Make code look sequential
  • ...which we may find easier to read

To Sum Up

Push & Pull

How to generate a series of asynchronous events?

  • Callbacks ties producer & consumer together
  • Promises resolve to a single value
  • Generators are synchronous-ish and don't multicast

 

 

 

 

One Pattern To Rule Them All

With Generators


export function interval(ms) {
  return eventChannel(emit => {
    const iv = setInterval(() => {
      emit(true);
    }, ms);

    return () => clearInterval(iv);
  });
}

export function* runTimer(id) {
  const intervalChannel = yield call(interval, 1000)

  try {
    while (true) {
      yield take(intervalChannel);
      yield fork(processInterval, id);
    }
  } finally {
    intervalChannel.close();
  }
}
                

export function* processInterval(id) {
  yield put(incrementTimer(id));
}

export function* manageTimer({ meta: { id } }) {
  const timerRunnerTask = yield fork(runTimer, id);

  yield take(action =>
    [STOP_TIMER, RESET_TIMER, DESTROY_TIMER]
        .includes(action.type)
        && action.meta.id === id
  );
  yield cancel(timerRunnerTask);
}

export function* watchStartTimer() {
  yield takeEvery(START_TIMER, manageTimer);
}

export default function* rootSaga() {
  yield fork(watchStartTimer);
}

With Observable


action$.whenAction(START_TIMER)
    .mergeMap(({meta}) => Observable.interval(1000).mapTo(meta.id))
    .map(id => incrementTimer(id))
    .takeUntil(action$.whenAction(STOP_TIMER, RESET_TIMER, DESTROY_TIMER))

KISS

...and don't make me think!

Best of both worlds?


action$.whenAction(LOAD_REQUEST)
    .mergeMap(() => Observable.spawn(function* () {
        yield loading();
        yield* fetchData().map(response => load(response))
        yield loaded();
    }))

Imperative programming

  • Code runs in a linear sequence
  • An instruction depends on previous execution
  • Need to enforce proper updating of dependants
  • Side-effects are natural
  • Idea: encapsulate side-effects

Functional ? Reactive Programming

What does that mean?

x = f(y) + g(x')
Reactive programming is programming with asynchronous data streams.
Values that change over time

Concepts

Streams

  • Ongoing events
  • Emits values to subscribers...
  • ...until it completes
  • ...or it signals an error

Streams: sources

  • Interactive events: mouse and keyboard input, timers
  • File system
  • Web service requests
  • Common infrastructure logic packaged as operators
  • Coordinate async calls (and cancellation/error handling)
  • Talk to observers whenever state changes
  • Start producing when a consumer subscribes

A Functional Flavour

  • Encapsulate low-level events into higher-level constructs
  • No visible mutation
  • First-class stream objects
  • Infinite, lazy evaluation
  • Reusable

Characteristics

  • Chain operators
  • Multiple subscribers
  • Retries on error signal
  • Schedulers parameterize concurrency

Create


const o = new Observable(observer => {
    fetchData((err, result) => {
        if (err) {
            observer.error(err)
        } else {
            observer.next(result)
            observer.complete()
        }
    })
})

Convert


Observable.of(value, ...)
Observable.from(promise/iterable/observable)

Compose


const mousedrag = mousedown.mergeMapTo(mousemove).takeUntil(mouseup)

Retry


const myObservable = obsA.mergeMap(a => getB(a).retry(3))
...

Subscription


const subscription = myObservable.subscribe(value => { ... })

Cancellation


subscription.unsubscribe()

Observables can be of any type


navigator.onLine
    ? Observable.timer(3000)
    : Observable.fromEvent(window, 'online').take(1)

Operators

map

It's interactive! Drag on any element!

scan

debounce

combineLatest

merge

zip

race

filter

distinct

pausable

One example

Typeahead

Whenever user types into a textbox, display a list of suggestions, but only if:

  • Text length is greater than 2 characters
  • User stopped typing for longer than 500ms
  • Text had actually been changed

Which event we're talking about?

The event `UserChangedTextLongerThan2CharactersAbout500msAgo`

When this event happens, make the request.

(and cancel any pending requests...)

One solution


const inputChanged = Observable.fromEvent(inputElement, 'input')
    .map(ev => ev.target.value)
    .filter(text => text.length > 2)
    .debounceTime(500)
    .distinctUntilChanged();
                    

const newSuggestionsArrived = inputChanged
    .switchMap(q => Observable.ajax.getJSON('search?term=' + q));
                    

newSuggestionsArrived.subscribe(data => {
    suggestionsElement.innerHTML = data.map(it => `
  • ${it}
    • `)
                                       .join('');
});

    Cold, Hot Observables

    • HOT: Shares a single producer for all subscribers
    • COLD: Creates a new producer for each subscriber

    Use cases

    Redux Store

    
function counter(state = 0, action) {
  switch (action.type) {
  case 'INCREMENT':
    return state + 1
  case 'DECREMENT':
    return state - 1
  default:
    return state
  }
}

const store = createStore(counter)

store.subscribe(state =>
  console.log(state)
)
                    
    
const actions$ = new SubjectBehavior()
const createStore = (reducer, preState) => action$.scan(reducer, preState)

    Side-effect model

    
action$
    .withLatestFrom(store)
    .filter(([action, state]) => action.type === 'INCREMENT_ODD')
    .filter(([action, state]) => state % 2 === 1)
    .map(() => {type: 'INCREMENT'})
    .subscribe(action$)
                    
    
action$
    .filter(action => action.type === 'LOAD_REQUEST')
    .mergeMap(() => fetchData().retry(3))
    .map(response => {type: 'LOAD_SUCCESS', payload: response})
    .catch(() => {type: 'LOAD_FAILED'})
    .subscribe(action$)

    Tips

    • Extract state and behaviors as streams
    • Bind them only once
    • Use a virtual time scheduler for testing

    Visualizing

    Implementations

    RxJS
    Bacon.js
    Cycle.js

    Future

    • Observable ES stage 1
    • Share a common push-based stream protocol

    And let's keep an eye on

    • Async generators
    • Communicating Sequential Processes

    Samples

    Thanks!

    Questions?

    @awerlang

    http://blog.werlangtecnologia.com.br

    References