Implement middleware

[gaearon]

Similar to https://github.com/rpominov/fluce/issues/4

[gaearon]

This is how Redux works now:

----[action]---->  [Redux] ----[state]---->
               (calls stores)

This is how I want it to work:

 ----[action]----> | ? | ----[action]---> [Redux] ----[state]----> | ? | ----[state]---->
                   | ? |                                           | ? |
                   | ? |                                           | ? |
                   | ? |                                           | ? |

Things in the middle are called interceptors. They are async black boxes. They are the extension points of Redux. There may be an interceptor that logs whatever goes through it. There may be an interceptor that transforms the values passing through it, or delays them. An interceptor should be able to fake the data, i.e. fire an arbitrary action at any point of time even if there's nothing being processed at the moment. This will be useful for time travel.

I think I found a nice API for interceptors. I'm going to rewrite Redux dispatcher to use them now. Even observing changes to stores can then be implemented as a (built-in) interceptor.

The interceptor signature is sink => payload => (). See the proof of concept below.

// Utilities
function noop() { }
function compose(...interceptors) {
  return sink => interceptors.reduceRight(
    (acc, next) => next(acc),
    sink
  );
}
function seal(interceptor) {
  return interceptor(noop);
}

// Dispatcher implementation
function getFirstAtom(stores) {
  return new Map([
    for (store of stores)
      [store, undefined]
  ]);
}
function getNextAtom(stores, atom, action) {
  if (!action) {
    return getFirstAtom(stores);
  }
  return new Map([
    for ([store, state] of atom)
      [store, store(state, action)]
  ]);
}
// Dispatcher is an interceptor too
function dispatcher(...stores) {
  stores = new Set(stores);
  let atom = getFirstAtom(stores);

  return sink => action => {
    atom = getNextAtom(stores, atom, action);
    sink(atom);
  };
}

// What can interceptors do?
// These are simple examples, but they show the power of interceptors.
function logger(label) {
  return sink => payload => {
    console.log(label, payload);
    sink(payload);
  };
}
function repeater(times) {
  return sink => payload => {
    for (let i = 0; i < times; i++) {
      sink(payload);
    }
  };
}
function delay(timeout) {
  return sink => payload => {
    setTimeout(() => sink(payload), timeout);
  };
}
function filter(predicate) {
  return sink => payload => {
    if (predicate(payload)) {
      sink(payload);
    }
  };
}
function faker(payload, interval) {
  return sink => {
    setInterval(() => sink(payload), interval);
    return noop;
  };
}

// Let's test this
import { counterStore } from './counter/stores/index';
import { todoStore } from './todo/stores/index';
import { increment } from './counter/actions/CounterActions';
import { addTodo } from './todo/actions/index';

let dispatch = seal(compose(
  // faker(increment(), 1000),
  // filter(action => action.type === 'INCREMENT_COUNTER'),
  // repeater(10),
  logger('dispatching action:'),
  dispatcher(counterStore, todoStore),
  // delay(3000),
  logger('updated state:')
));

dispatch({ type: 'bootstrap' });
dispatch(increment());

setTimeout(() => {
  dispatch(increment());
  dispatch(addTodo());
}, 1000);

[rpominov]

Pretty solid, I like it. My approach looks kind of "dirty" compared to it, but I think it a bit more practical. Two notable differences from what I'm going to do in fluce:

1. You have two points where programmer can insert an interceptor — before and after dispatcher. Maybe there will be difficulty with API, maybe not — don't know )
2. In fluce the reducer function will be available in middleware, and here only dispatcher will have access to such reducer. I used reducer in transactions-middleware, so it seems necessary to have access to it in middleware/interceptor.

[gaearon]

Yeah I'll need to battle test it. Maybe need to make it more practical, maybe not :-) Transactions are a good testbed indeed.

[gaearon]

You have two points where programmer can insert an interceptor — before and after dispatcher.

I think the interceptors I described here are too low-level. In practice, each interceptor should wrap some other interceptor. This way it's possible for each interceptor to add intercepting functions before and after the nested interceptor.

I used reducer in transactions-middleware, so it seems necessary to have access to it in middleware/interceptor.

Indeed, my proposal needs to be changed to support transactions. It might be that the API is enough, but the dispatcher should accept state, actions as the payload, and output state. Dispatcher should not store the state. This goes well with “storing state is just an optimization, in theory it should be equivalent to reducing all the actions”. The state memoization could then be implemented as the outermost interceptor.

Hopefully this doesn't sound too crazy.. I'll try to get something running tomorrow.

[rpominov]

It might be that the API is enough, but the dispatcher should accept state, actions as the payload, and output state.

(state, actions) -> state looks like the reducer function only accepts multiple actions. Or did you mean that the dispatcher outputs the state by calling a callback rather than simply return it?

Anyway, looks promising. Will be interesting to see more details tomorrow.

[KyleAMathews]

Why not middleware just (conceptually) be another store? Why a new concept for interceptors?

Per the gist discussion, if stores can compose stores then middleware is just an intermediate store?

[rpominov]

It won't be powerful enough as another store. A middleware should be able to fire actions, replace current state etc.

[gaearon]

For extra clarification, the middleware is not meant to be used by the lib users. It is meant for tool builders (e.g. DevTools extensions).

[KyleAMathews]

:+1:

[gaearon]

(state, actions) -> state looks like the reducer function only accepts multiple actions. Or did you mean that the dispatcher outputs the state by calling a callback rather than simply return it?

It may simply return it. Seems like this signature enables transactions with no need for async-y things (which is a huge relief). Such dispatcher might look like this:

export function dispatcher(...stores) {
  function init() {
    return new Map([
      for (store of stores)
        [store, undefined]
    ]);
  }

  function step(atom, action) {
    return new Map([
      for ([store, state] of atom)
        [store, store(state, action)]
    ]);
  }

  return function dispatch(atom = init(), actions = []) {
    return actions.reduce(step, atom);
  };
}

[rpominov]

Yeah, but you still need to allow asynchrony in interceptors, so they could delay dispatches, dispatch at any time, etc.

If interceptors will wrap each other, I guess they will be functions that accepts some object and return an object of the same shape. The question is what is the shape of that object will be? It can't be (state, actions) -> state function, because we need asynchrony, but such function should be part of that object (because it needed). So perhaps what I was proposing is the way after all https://github.com/rpominov/fluce/issues/4#issuecomment-107402900

[gaearon]

@rpominov

I'm not sure async is needed at this point.
I got transactions working. Take a look at what I got.

• memoize, replay and transact are different dispatch strategies. You may use either.
• No transaction.begin() etc. Everything is action, even a meta action.

Example:

import { reducer, log, memoize, replay, transact } from './redux';
import { counterStore } from './counter/stores/index';
import { todoStore } from './todo/stores/index';
import { increment } from './counter/actions/CounterActions';

let reduce = reducer(counterStore, todoStore);

console.log('--------- naive ---------');
let naiveDispatch = log(replay(reduce));
naiveDispatch(increment());
naiveDispatch(increment()); // will call stores twice
naiveDispatch(increment()); // will call stores three times

console.log('--------- caching ---------');
let cachingDispatch = log(memoize(reduce));
cachingDispatch(increment());
cachingDispatch(increment()); // will call store just once
cachingDispatch(increment()); // will call store just once

console.log('--------- transactions ---------');
let dispatch = log(transact(reduce));
dispatch(increment());
dispatch(increment());

dispatch(transact.BEGIN); // lol I'm dispatching a built-in action!
dispatch(increment());
dispatch(increment());

dispatch(transact.ROLLBACK); // lol I'm dispatching a built-in action!
dispatch(increment());
dispatch(increment());

dispatch(transact.COMMIT); // lol I'm dispatching a built-in action!
dispatch(increment());

Impl:

// Type definitions:
// reduce: (State, Array<Action>) => State
// dispatch: (State, Action) => State

// --------------------------
// Reducer (stores -> reduce)
// --------------------------

export function reducer(...stores) {
  function init() {
    return new Map([
      for (store of stores)
        [store, undefined]
    ]);
  }

  function step(state, action) {
    return new Map([
      for ([store, slice] of state)
        [store, store(slice, action)]
    ]);
  }

  return function reduce(state = init(), actions) {
    return actions.reduce(step, state);
  };
}

// -------------------
// Dispatch strategies
// (reduce -> dispatch)
// -------------------

export function memoize(reduce) {
  let state;

  return function dispatch(action) {
    state = reduce(state, [action]);
    return state;
  };
}

export function replay(reduce) {
  let actions = [];

  return function dispatch(action) {
    actions.push(action);
    return reduce(undefined, actions);
  };
}

export function transact(reduce) {
  let transacting = false;
  let committedState;
  let stagedActions = [];
  let state;

  return function dispatch(action) {
    switch (action) {
    case transact.BEGIN:
      transacting = true;
      stagedActions = [];
      committedState = state;
      break;
    case transact.COMMIT:
      transacting = false;
      stagedActions = [];
      committedState = state;
      break;
    case transact.ROLLBACK:
      transacting = false;
      stagedActions = [];
      state = committedState;
      break;
    default:
      if (transacting) {
        stagedActions.push(action);
        state = reduce(committedState, stagedActions);
      } else {
        state = reduce(state, [action]);
        committedState = state;
      }
    }
    return state;
  };
}
transact.BEGIN = { type: Symbol('transact.BEGIN') };
transact.COMMIT = { type: Symbol('transact.COMMIT') };
transact.ROLLBACK = { type: Symbol('transact.ROLLBACK') };

// ----------------------
// Wrappers
// (dispatch -> dispatch)
// ----------------------

export function log(dispatch) {
  return (action) => {
    console.groupCollapsed(action.type);
    const state = dispatch(action);
    console.log(state);
    console.groupEnd(action.type);
    return state;
  };
}

[rpominov]

Cool :+1:

To use actions to control interceptors is smart! Also you can still implement async stuff (like delay) using "Wrappers", if I understand right.

But you can't use more than one "Dispatch strategy", right? Perhaps it not needed, but still...

[gaearon]

Edit: fixed some bugs in transact implementation.

[gaearon]

Yeah, can't use more than one. I couldn't find a way for many to make sense though. Somebody has to own the state.

[gaearon]

To use actions to control interceptors is smart!

This is my favorite part. It's the only sane answer I found to “how does an interceptor initiate changes when it's in a middle of a chain?”

[ooflorent]

Let's add async actions and your dispatcher will be awesome!

[gaearon]

@ooflorent What do you mean by async actions? If you're speaking of built-in Promise support, I think this could too be implemented as a middleware.

[rpominov]

You may also want to consider optimistic dispatches as a battle test for the API. Maybe this is the case when we might want two dispatch strategies at once (e.g. optimistic dispatches and transactions).

[gaearon]

Good point. I'll try that. (Also async actions, and observing stores.)

[gaearon]

I thought about it some more. I'm not sure that aiming for very composable middleware makes sense, at least for me at this stage. It's hard to say, for example, how transactions could work together with time travel. I'm not convinced it's feasible or desirable to implement them separately. Therefore the concept of a single "dispatch strategy" might be enough for my goals, at least for now.

As for optimistic dispatch, I'm not sure I want to support this as a middleware. This sounds like something that can be done at the store level, potentially with a shared utility. The middleware would alter state of the world (remove actions that have "happened") which feels non-Flux. I'm okay with DevTools doing this, but I'd rather stick to Flux way for anything that would be used in production.

[rpominov]

Makes sense.

And yeah, I also have a controversial feeling about optimistic dispatches. It seems so "cool" to be able to remove actions from the history, but also looks like a good footgun. Not sure I would use it in production.

[gaearon]

Here's my next stab. It differs in scope from what was proposed in #55. The first comments are concerned with perform strategy. I'm going to describe a draft of a possible solution to solving dispatch strategies. I'm not convinced it's the same problem, just yet. I'm also not convinced that #55 could solve dispatch strategies without introducing asynchrony inside dispatch. I don't want asynchrony inside dispatch. (see below)

---

Sorry for crazy Greek letters. I'm not high or anything..

This is proof of concept of “shadow Flux”. I'll write up something better after I figure out how to compose these things. The idea is to use the same Flux workflow for the middleware. No asynchrony. The middleware's API is like Stores, but operating on higher-level entities.

• Σ is like “lifted state”. It has real state as a field.
• Δ is like “lifted action”. It has real action as a field.

The middleware API is (Σ, Δ) -> Σ. Kinda looks like Store, haha (intentional). It is probably possible to compose the middleware exactly the same way we can compose our Stores.

Middleware doesn't keep any state in closures. Instead, it acts exactly as Redux Stores.

There is one built-in “lifted action”: RECEIVE_ACTION. It is fired whenever a real action is fired. The real action is in its action field.

import counter from './stores/counter';
import { increment } from './actions/CounterActions';

const RECEIVE_ACTION = Symbol();

function liftAction(action) {
  return { type: RECEIVE_ACTION, action };
}

Here are a few naive middlewares:

// ---------------
// Calculates state by applying actions one by one (DEFAULT!)
// ---------------
(function () {
  function memoizingDispatcher(reducer) {
    const Σ0 = {
      state: undefined
    };

    return (Σ = Σ0, Δ) => {
      switch (Δ.type) {
      case RECEIVE_ACTION:
        const state = reducer(Σ.state, Δ.action);
        return { state };
      }
      return Σ;
    };
  }

  let dispatch = memoizingDispatcher(counter);
  let nextΣ;

  for (let i = 0; i < 5; i++) {
    nextΣ = dispatch(nextΣ, liftAction(increment()));
    console.log(nextΣ);
  }
})();

// ---------------
// Calculates state by replaying all actions over undefined atom (not very practical I suppose)
// ---------------
(function () {
  function replayingDispatcher(reducer) {
    const Σ0 = {
      actions: [],
      state: undefined
    };

    return (Σ = Σ0, Δ) => {
      switch (Δ.type) {
      case RECEIVE_ACTION:
        const actions = [...Σ.actions, Δ.action];
        const state = actions.reduce(reducer, undefined);
        return { actions, state };
      }
      return Σ;
    };
  }

  let dispatch = replayingDispatcher(counter);
  let nextΣ;

  for (let i = 0; i < 5; i++) {
    nextΣ = dispatch(nextΣ, liftAction(increment()));
    console.log(nextΣ);
  }
})();

Any middleware may handle other “lifted actions” defined just by it. For example, my custom gateDispatcher defines LOCK_GATE and UNLOCK_GATE:

// --------------
// This is like a watergate. After LOCK_GATE, nothing happens.
// UNLOCK_GATE unleashes accumulated actions.
// --------------
(function () {
  const LOCK_GATE = Symbol();
  const UNLOCK_GATE = Symbol();

  function gateDispatcher(reducer) {
    const Σ0 = {
      isLocked: false,
      pendingActions: [],
      lockedState: undefined,
      state: undefined
    };

    return (Σ = Σ0, Δ) => {
      switch (Δ.type) {
      case RECEIVE_ACTION:
        return {
          isLocked: Σ.isLocked,
          lockedState: Σ.lockedState,
          state: Σ.isLocked ? Σ.lockedState : reducer(Σ.state, Δ.action),
          pendingActions: Σ.isLocked ? [...Σ.pendingActions, Δ.action] : Σ.pendingActions
        };
      case LOCK_GATE:
        return {
          isLocked: true,
          lockedState: Σ.state,
          state: Σ.state,
          pendingActions: []
        };
      case UNLOCK_GATE:
        return {
          isLocked: false,
          lockedState: undefined,
          state: Σ.pendingActions.reduce(reducer, Σ.lockedState),
          pendingActions: []
        };
      default:
        return Σ;
      }
    };
  }

  let dispatch = gateDispatcher(counter);
  let nextΣ;

  for (let i = 0; i < 5; i++) {
    nextΣ = dispatch(nextΣ, liftAction(increment()));
    console.log(nextΣ);
  }

  nextΣ = dispatch(nextΣ, { type: LOCK_GATE });

  for (let i = 0; i < 5; i++) {
    nextΣ = dispatch(nextΣ, liftAction(increment()));
    console.log(nextΣ);
  }

  nextΣ = dispatch(nextΣ, { type: UNLOCK_GATE });
  console.log(nextΣ);
})();

Why this is cool:

• No asynchrony. sink => action API sucks because of asynchrony. Async is hard.
• No closured state, thus composable. Just like Stores. “Parent” middleware can manage “child”.
• Deterministic. Just like normal Flux, “every change is caused by action”. Only, a “lifted action” in this case.

Open questions:

• How exactly to compose these folks
• Explain how this solves my problem better than #55 (I'll try after I have some success with composing)

[gaearon]

Here's an example of composition:

import counter from './stores/counter';
import { increment } from './actions/CounterActions';

const RECEIVE_ACTION = Symbol();

function liftAction(action) {
  return { type: RECEIVE_ACTION, action };
}

// ---------------------------

(function () {
  function replay(reducer) {
    const Σ0 = {
      actions: [],
      state: undefined
    };

    return (Σ = Σ0, Δ) => {
      switch (Δ.type) {
      case RECEIVE_ACTION:
        const actions = [...Σ.actions, Δ.action];
        const state = actions.reduce(reducer, undefined);
        return { actions, state };
      }
      return Σ;
    };
  }

  const LOCK_GATE = Symbol();
  const UNLOCK_GATE = Symbol();

  function gate(next) { // Note: instead of `reducer`, accept `next`
    const Σ0 = {
      isLocked: false,
      pendingActions: [],
      lockedState: undefined,
      state: undefined
    };

    return (Σ = Σ0, Δ) => {
      switch (Δ.type) {
      case RECEIVE_ACTION:
        return {
          isLocked: Σ.isLocked,
          lockedState: Σ.lockedState,
          state: Σ.isLocked ? Σ.lockedState : next(Σ.state, Δ),
          pendingActions: Σ.isLocked ? [...Σ.pendingActions, Δ.action] : Σ.pendingActions
        };
      case LOCK_GATE:
        return {
          isLocked: true,
          lockedState: Σ.state,
          state: Σ.state,
          pendingActions: []
        };
      case UNLOCK_GATE:
        return {
          isLocked: false,
          lockedState: undefined,
          state: Σ.pendingActions.map(liftAction).reduce(next, Σ.lockedState),
          pendingActions: []
        };
      default:
        return Σ;
      }
    };
  }

  let dispatch = gate(replay(counter)); // Gate is outermost. When Gate is unlocked, Replay is used.
  let nextΣ;

  nextΣ = dispatch(nextΣ, { type: LOCK_GATE });
  console.log(nextΣ);

  for (let i = 0; i < 5; i++) {
    nextΣ = dispatch(nextΣ, liftAction(increment()));
    console.log(nextΣ);
  }

  nextΣ = dispatch(nextΣ, { type: UNLOCK_GATE });
  console.log(nextΣ);
})();

[rpominov]

I want to try to summarize what we have. Hopefully this will help.

Here is the beast we try to insert middleware in:

It has something with a direct access to the state atom, when we dispatch an action, it

1. takes the current state and the action,
2. calls reducer with that to get a new state,
3. and replaces current state with a new one.

---

We have 4 meaningful insertion points here. We can insert functions (...args) => next(...args) in following places:

// #1
function(state, action) {
  next(state, action)
}

// #2
function(state) {
  next(state)
}

// #3
function(action) {
  next(action)
}

Also in points 2 and 3 we can insert mapping functions (state) -> state and (action) -> action, but they are not very useful compared to functions that call next.

The middleware for point 1-3 looks like this:

// Takes a next function and returns a new next function
function(next) {
  return (...args) => {
    next(...args)
  }
}

Also we can wrap whole reducer:

The middleware for point 4 looks like this:

// Takes a reducer and returns a new one
function(reducer) {
  return (state, action) => reducer(state, action)
}

---

Both formats of middleware (for points 1-3, and for point 4) naturally composable i.e., we can compose several middlewares using simple compose function. But let look what happens when we compose them. Suppose we did compose(a, b, c) for each point:

Notice the order at which they will be applied, it might be important. Of course if we do composeRight instead of compose the order will be opposite.

---

Now let try to classify all proposed APIs.

1. 55 proposes to use insertion point 3 (with getAtom() thingie)

2. https://github.com/gaearon/redux/issues/6#issuecomment-109446943 proposes to use point 4, but in a bit different way than (reducer) -> reducer, but I think it opens same opportunities as (reducer) -> reducer (also the different way isn't composable)
3. https://github.com/gaearon/redux/issues/6#issuecomment-109768420 Also variation of point 4? Fairly I can't tell. @gaearon 's proposals don't fit well to this "theory" of mine :)
4. https://github.com/rpominov/fluce/issues/4#issuecomment-107402900 proposes to use points 1, 2, and 4 together. But we should be careful with order of composition, we should apply compose() to some points and composeRight() to others, but I don't sure which ones.

---

Edit: note about getAtom() added.

[rpominov]

Actually, I think I'm wrong that #55 simply uses insertion point 3. Because "something that holds the state" is just another middleware in it.

Looks like this "theory" describes only my view on the problem...

cc @acdlite

[gaearon]

Wow, these are neat! I think you're missing middleware API calls. I model them as "lifted" actions, of which normal actions are a subset. Otherwise you have to assume any middleware can initiate state changes while being "behind" other middleware. I think this is what's problematic.

I'll try to come up with better code and explanations for my last example.

[rpominov]

Yeah, I see what you're doing with "lifted" actions. You use point 4, which means middlewares are synchronous. But you still can replace state at arbitrary time using controlling actions. And it also composable now, so it actually looks great! :+1:

[gaearon]

I want to start by making it possible to implement any middleware solution in the userland: https://github.com/gaearon/redux/pull/60

[gaearon]

60 now provides an extensibility point to implement any kind of middleware system. I'm closing this issue for now. We'll probably revisit it later after learning to write custom dispatchers and taking some lessons from that.