Redux with Rxjs

index.js

"use strict";
const Rx = require('rx');
const fetch = require('isomorphic-fetch'); /* use the fetch api on client and server */

/**
 * Given a subreddit, pull down post ids.  that is, changing the subreddit by calling terms$.onNext(SUBREDDITNAME)
 * automatically calls the reddit api and populates the store.
 * To try it out, npm install rx and isomorphic-fetch, then
 * var S = require('./index.js');
 * S.store$.subscribe(x => console.log(x));  // listen to every state change
 * S.terms$.onNext('javascript');
 */

/**
 * Constants
 */
const RECIEVE_POSTS = 'RECIEVE_POSTS';
const UPDATE_SUBREDDIT = 'UPDATE_SUBREDDIT';

/**
 * Reducer
 */
const reducer = (state, action) => {
  switch(action.type) {
  case UPDATE_SUBREDDIT:
    return Object.assign({}, state, {
      term: action.payload.term,
      isLoading: true
    });
  case RECIEVE_POSTS:
    return Object.assign({}, state, {
      posts: action.payload.posts,
      isLoading: false
    });
  default:
    return state;
  }
}

/**
 * Actions - Recievers
 * posts$ - recieves JSON response from wikipedia api
 * terms$ - recieves string from input
 * we can provide data to these by calling `.onNext(x)` or in 5-beta just `.next(x)`
 */
const posts$ = new Rx.Subject();
const terms$ = new Rx.Subject();


/**
 * Wiring - this simply combines our actions into our store a la redux
 */
const initialState = {
  posts: [],
  term: '',
  isLoading: false
};
const store$ = Rx.Observable
  .merge(
    posts$.map(posts => ({ type: RECIEVE_POSTS, payload: { posts }})),
    terms$.map(term => ({ type: UPDATE_SUBREDDIT, payload: { term }}))
  )
  .startWith(initialState)
  .scan(reducer);

/**
 * Actions - subscribers
 * fetch new posts when terms changes
 */

// helper function for reddit call
const api = (term) => {
  return fetch(`https://www.reddit.com/r/${term}.json`, {method: 'GET'})
    .then(d => d.json());
};

/**
 * Listen to the store, when store.term changes, call the api and refresh posts
 */
store$
  .map(store => store.term)
  .distinctUntilChanged() /* Don't do anything if store.term doesn't change */
  .flatMapLatest(term => Rx.Observable.fromPromise(api(term)))  /* cancel previous calls if new one comes in */
  .forEach(posts => {  /* call appropriate action */
    posts$.onNext(posts.data.children.map(d => d.data.id));
  });

module.exports = {
  store$,
  posts$,
  terms$
};