The Demo GIF will look like this:

The basic idea

Since the redux change data is dispatches (actions), it is natural to use the action as the base unit to transmit between client and client, and to store the action in arrays on client and server. As long as the order of the action queues on client and server remains the same, Reducer is a pure function and we know that the calculated state is the same.

Some conventions

In this paper, C1, C2… Cn indicates the client, S indicates the server, A1, A2, A3… An stands for aciton, and the server side is written using KOA + socket. IO (as a front-end, the server side has almost zero knowledge).

The overall train of thought

When a client initiates an action and the server receives the action, the server does three things:

  1. Push the action onto the stack
  2. Send the action before the client a1 to the client (like git pull before git push)
  3. Send A1 to other clients

However, the above idea is more general, think carefully will find many problems:

  1. What if C2, C3, and Action are distributed when A1 reaches S?
  2. What if C1 is receiving actions sent by other clients when A1 arrives at S?
  3. What if C1 is sending the previous action before A1 sends it? We’ll solve them one by one.

The server sends the action mechanism

The server has two concepts: target, member, edit object (report, flow chart, etc.), and edit user. When you send two actions: In the case of A1 and A2, due to the uncertainty of network transmission sequence, a1 should be sent first, and then A2 should be sent after the client receives the packet. In this way, the sequence of A1 and A2 can be guaranteed on the client. Therefore, each member has a pending variable indicating whether the action is being sent or not, and index indicating the index of the latest action that was sent in the action queue.

When a server receives aciton from a client

this.socket.on('client send action', (action) => {
    // The target queues the action and returns its index
    let index = this.target.addAction(action);
      if (this.pending) {
        this.queue.push({
          method: 'sendBeforeActions'.args: [index]
        })
      } else {
        this.sendBeforeActions(index);
      }
      this.target.receiveAction({
        member: this,
        index
      })
    })
Copy the code

These are the three things that the server does when it receives a1. If so, add aciton before sending A1 to an action queue and tell target that this member sent an action.

sendBeforeActions

sendBeforeActions(refIndex) {
    let {actions} = this.getCurrent(refIndex);
    actions = actions.slice(0.- 1);
    this.pending = true;
    this.socket.emit('server send before actions', { actions }, () => {
      this.pending = false;
      this.target.setIndex(this, refIndex);
      this.sendProcess();
    });
  }
Copy the code

This function receives an index, which in the above code is the index of the actions received by the member in the queue, so getCurrent(refIndex) refers to all actions that haven’t been sent to the member by refIndex (which may be empty), So actions. Slice (0, -1) is sent to the client after culling itself. Terminates the send status, sets the index of member’s latest action, and then calls sendProcess to see if any subsequent actions were queued during its own sending

  sendProcess() {
    if (this.queue.length > 0&&!this.pending) {
      let current = this.queue.shift();
      let method = this[current.method];
      method.apply(this, current.args); }}Copy the code

If you noticed that:

if (this.pending) {
    this.queue.push({
        method: 'sendBeforeActions',
        args: [index]
    })
}
Copy the code

You’ll notice that if the member is sending another action when it wanted to send before action, it will wait for the action to send before triggering sendProcess to send.

Also send this action to other users

In that code

// This refers to a member object
this.target.receiveAction({
    member: this,
    index
})

Copy the code

This is what triggers the other client to send

ReceiveAction ({member, index}) {this.members. ForEach (m => {if(m.id ! == member.id) { m.queue.push({ method:'sendActions', args: [index] }); m.sendProcess(); }})}Copy the code

If members has a member of the sender, the send action is put in the member send queue and sendProcess is executed

sendActions

 sendActions(refIndex) {
    let {actions} = this.getCurrent(refIndex);
    if (actions.length) {
      this.pending = true;
      this.socket.emit('server send actions', {actions}, () => {
        this.pending = false; this.target.setIndex(this, refIndex); this.sendProcess(); }}})Copy the code

This function is almost the same as sendBeforeActions, except to discard the latest actions, thus ensuring the order in which the server sends the actions

Client IO middleware

On the client side, IO related operations are encapsulated in a middleware

module.exports = store => next => action => {
    if (action.type === 'connection') {// The connection initializes some eventsreturn initIo(action.payload)
    }  
    if (action.type === 'disconnection') {
        return socket.disconnect(action.payload)
    }
    if (['@replace/state'].indexOf(action.type.toLowerCase()) === -1 && ! action.escapeServer && ! Action. Temporary) {// Set action to userId, targetId Action = actionCreator(action); // Get the new action queue, calculate the actions, and update to stateletnewCacheActions = [...cacheActions, action]; mapActionsToState(newCacheActions); // Send to the serverreturndelieverAction(action); } // This allows only replace state actions to be stored in the store. This is a requirement for implementing undo/redo, which I'll cover next(); }Copy the code

Some global variables

We’ll use that later

letcacheActions = []; // Action queue, which is consistent with the action queue on the server sideletcurrentActions = []; // Actions calculated according to cacheActionsletredoActions = {}; // Cache each user's undo and remove the actionlet pending = false; // Whether the request is being sentletactionsToPend = []; // Cache send queueletbeforeActions = []; // Cache the actions pulled downletcurrentAction = null; // Action currently sentletuser, tid; // User name and targetIdletinitialState; // Initial statelettimeline = []; / / the cache stateCopy the code

The overall thinking diagram of the client

There are two main places:

(1) During computeActions, the reason why undo removes the last action of the user and promotes the penultimate action to the last is that if there are other user actions occurring after the penultimate action of the user, Then, other users may overwrite the set value of the user action, and the user cannot return to the previous state when undo. In this case, the promotion is equivalent to making a new action after undo, and this action is the previous action. This algorithm is buggy. When a user undoes, the change of the action that conflicts with the action will be overwritten because we will promote the next-to-last action. There is something wrong with this conflict-resolution strategy. If not promoted, the user cannot undo if his previous action was overwritten by another user’s action. This is a pain point, I still continue to explore, welcome god’s advice.

(2) The user received actions while pending, which is equivalent to before Actions. The code for the main functions is posted below

initIo

function initIo(payload, dispatch) {
  user = payload.user;
  tid = parseInt(payload.tid, 10);
  // Initialize the socket
  let socket = cache.socket = io(location.protocol + '/ /' + location.host, {
    query: {
      user: JSON.stringify(user),
      tid
    }
  });
  // Get the initial data
  socket.on('deliver initial data', (params) => { ... Get initial state, actions})// Sending an action waits for the actions before pulling
  socket.on('server send before actions', (payload, callback) => {
    pending = false;
    callback && callback();
    let {actions} = payload;
    actions = [...actions, ...beforeActions, currentAction];
    cacheActions = [...cacheActions, ...actions];
    if (actions.length > 1) {
      // Verify that the previous action exists, and recalculate the state from the actions
      mapActionsToState();
    }
    if (actionsToPend.length) {
      letaction = actionsToPend.shift(); sendAction(action); }})/ / receive the actions
  socket.on('server send actions', (payload, callback) => {
    let {actions} = payload;
    callback && callback();
    if (pending) {
      beforeActions = [...beforeActions, ...actions];
    } else{ cacheActions = [...cacheActions, ...actions]; mapActionsToState(); }})}Copy the code

mapActionsToState

function mapActionsToState(actions) {
  actions = actions || cacheActions;
  if (actions.length === 0) {
    return replaceState(dispatch)(initialState);
  }
  let {newCurrentActions, newRedoActions} = computeActions(actions);
  let {same} = diff(newCurrentActions);

  let state = initialState;
  if (timeline[same]) {
    state = timeline[same];
    timeline = timeline.slice(0, same + 1);
  }
  if (same === - 1) {
    timeline = [];
  }
  let differentActions = newCurrentActions.slice(same + 1);
  differentActions.forEach(action= > {
    state = store.reducer(state, action);
    timeline.push(state);
  });
  currentActions = newCurrentActions;
  redoActions = newRedoActions;
  store.canUndo = (a)= > currentActions.some(action= > action.userId === user.id);
  store.canRedo = (a)= >!!!!! (redoActions[user.id] || []).length;return replaceState(dispatch)(state);
}
Copy the code

computeActions

function computeActions(actions) {
  let newCurrentActions = [];
  let newRedoActions = {};
  actions.forEach(action= > {
    let type = action.type.toLowerCase();
    newRedoActions[action.userId] = newRedoActions[action.userId] || [];
    if(type ! = ='redo'&& type ! = ='undo') {
      newCurrentActions.push(action);
      newRedoActions[action.userId] = [];
    }
    if (type === 'undo') {
      let indexes = [];
      for (let i = newCurrentActions.length - 1; i >= 0; i--) {
        if (newCurrentActions[i].userId === action.userId) {
          indexes.push(i);
        }
        if (indexes.length === 2) {
          break; }}if (indexes.length > 0) {
        let redo = newCurrentActions.splice(indexes[0].1) [0];
        newRedoActions[action.userId].push(redo);
      }
      if (indexes.length > 1) {
        let temp = newCurrentActions.splice(indexes[1].1);
        newCurrentActions.push(temp[0]); }}if (type === 'redo') {
      letredo = newRedoActions[action.userId].pop(); newCurrentActions.push(redo); }});return {
    newCurrentActions,
    newRedoActions
  }
}
Copy the code

diff

function diff(newCurrentActions) {
  let same = - 1;
  newCurrentActions.some((action, index) = > {
    let currentAction = currentActions[index];
    if (currentAction && action.id === currentAction.id) {
      same = index;
      return false;
    }
    return true;
  });
  return {
    same
  }
}
Copy the code

conclusion

Speak a lot, do not know to have to spell out of his thoughts, in which their demo also run up, only had two browsers to simulate testing, always feel some concurrent delay will have bugs, behind will continue to optimize the idea, add some automated tests to verify, in addition, also haven’t consider to server storage, at first only to run in memory, Will think about preservation plan. I hope those who are interested in this aspect can give me some guidance