rxdux-rx

async.js

const Rx = require('rxjs');

const sleep = ms => new Promise(res => setTimeout(res, ms));

const action$ = new Rx.Subject();

const initialState = {
  bays: 3
};

const reducer = (state, action) => {
  // console.log({state, action})
  switch(action.type) {
    case 'CHANGE_BAYS':
      return {
        ...state,
        bays: action.payload
      };
    default:
      return state;
  }
}

const state$ = action$
  .switch()
  .startWith(initialState)
  .scan(reducer)

const dispatch = func => args => {
  if (func[Symbol.toStringTag] === 'AsyncFunction') {
    action$.next(Rx.Observable.fromPromise(func(args)));
  } else {
    action$.next(Rx.Observable.of(func(args)));
  }
}

const addBaysAsync = async function(payload) {
  await sleep(1000);
  return {
    type: 'CHANGE_BAYS',
    payload
  }
};

const addBays = function(payload) {
  return {
    type: 'CHANGE_BAYS',
    payload
  }
};

state$.subscribe(s => console.log({s}));

dispatch(addBaysAsync)(4);
dispatch(addBays)(1);
dispatch(addBaysAsync)(20);
dispatch(addBaysAsync)(2);

async_with_loading_state.js

const { Observable, Subject } = require('rxjs');

const sleep = ms => new Promise(res => setTimeout(res, ms));

const action$ = new Subject();

const initialState = {
  bays: 3
};

const reducer = (state, action) => {
  // console.log({state, action})
  switch(action.type) {
    case 'CHANGE_BAYS':
      return {
        ...state,
        changing: true
      };
    case 'CHANGED_BAYS':
      return {
        ...state,
        changing: false,
        bays: action.payload
      };
    default:
      return state;
  }
}

const state$ = action$
  .switch()
  .startWith(initialState)
  .scan(reducer)

const dispatch = func => args => {
  const action = func(args);
  action$.next(Observable.of(action));
  if (action.payload[Symbol.toStringTag] === 'AsyncFunction') {
    action$.next(Observable.fromPromise(action.payload(args)));
  }
}

const addBaysAsync = function(payload) {
  return {
    type: 'CHANGE_BAYS',
    payload: async function(payload) {
      await sleep(1000);
      return {
        type: 'CHANGED_BAYS',
        payload
      }
    }
  }
};

const addBays = function(payload) {
  return {
    type: 'CHANGED_BAYS',
    payload
  }
};

state$.subscribe(s => console.log({s}));

dispatch(addBaysAsync)(4);
// dispatch(addBays)(1);
dispatch(addBaysAsync)(20);

sync.js

const Rx = require('rxjs');

const action$ = new Rx.Subject();

const initialState = {
  bays: 3
};

const reducer = (state, action) => {
  switch(action.type) {
    case 'CHANGE_BAYS':
      return {
        ...state,
        bays: action.payload
      };
    default:
      return state;
  }
}

const state$ = action$
  .startWith(initialState)
  .scan(reducer);

const dispatch = func => (...args) => {
  action$.next(func(...args));
}

const addBays = function(payload) {
  return {
    type: 'CHANGE_BAYS',
    payload
  }
};

state$.subscribe(s => console.log(s));

dispatch(addBays)(1);
dispatch(addBays)(2);