Improve flush logic

Author: christianalfoni

packages/app/src/app/overmind/effects/live/clients.ts

@@ -1,4 +1,4 @@
-import { Client } from 'ot';
+import { Client, TextOperation } from 'ot';
 
 export type SendOperation = (
   moduleShortid: string,
@@ -8,24 +8,11 @@ export type SendOperation = (
 
 export type ApplyOperation = (moduleShortid: string, operation: any) => void;
 
-function operationToElixir(ot) {
-  return ot.map(op => {
-    if (typeof op === 'number') {
-      if (op < 0) {
-        return { d: -op };
-      }
-
-      return op;
-    }
-
-    return { i: op };
-  });
-}
-
 class CodeSandboxOTClient extends Client {
   moduleShortid: string;
   onSendOperation: (revision: string, operation: any) => void;
   onApplyOperation: (operation: any) => void;
+  saveOperation: TextOperation;
 
   constructor(
     revision: number,
@@ -39,28 +26,40 @@ class CodeSandboxOTClient extends Client {
     this.onApplyOperation = onApplyOperation;
   }
 
-  sendOperation(revision, operation) {
-    this.onSendOperation(revision, operationToElixir(operation.toJSON()));
-  }
+  flush() {
+    const saveOperation = this.saveOperation;
 
-  applyOperation(operation) {
-    this.onApplyOperation(operation);
+    this.saveOperation = null;
+
+    return {
+      revision: this.revision,
+      operation: saveOperation,
+    };
   }
 
-  serverAck() {
-    super.serverAck();
+  revertFlush(operation) {
+    if (this.saveOperation) {
+      this.saveOperation = operation.compose(this.saveOperation);
+    } else {
+      this.saveOperation = operation;
+    }
   }
 
-  applyClient(operation: any) {
-    super.applyClient(operation);
+  sendOperation(revision, operation) {
+    this.onSendOperation(revision, operation);
   }
 
-  applyServer(operation: any) {
-    super.applyServer(operation);
+  applyOperation(operation) {
+    this.onApplyOperation(operation);
   }
 
-  serverReconnect() {
-    super.serverReconnect();
+  applyClient(operation: any) {
+    if (this.saveOperation) {
+      this.saveOperation = this.saveOperation.compose(operation);
+    } else {
+      this.saveOperation = operation;
+    }
+    super.applyClient(operation);
   }
 }
 

packages/app/src/app/overmind/effects/live/index.ts

@@ -7,15 +7,16 @@ import {
 import _debug from '@codesandbox/common/lib/utils/debug';
 import { camelizeKeys } from 'humps';
 import { TextOperation } from 'ot';
-import { Socket, Channel } from 'phoenix';
+import { Channel, Socket } from 'phoenix';
 import uuid from 'uuid';
 
-import clientsFactory from './clients';
 import { OPTIMISTIC_ID_PREFIX } from '../utils';
+import clientsFactory from './clients';
 
 type Options = {
   onApplyOperation(args: { moduleShortid: string; operation: any }): void;
   provideJwtToken(): string;
+  getConnectionsCount(): number;
 };
 
 type JoinChannelResponse = {
@@ -29,27 +30,22 @@ declare global {
   }
 }
 
-const identifier = uuid.v4();
-const sentMessages = new Map();
-const debug = _debug('cs:socket');
+class Live {
+  private identifier = uuid.v4();
+  private sentMessages = new Map();
+  private debug = _debug('cs:socket');
+  private channel: Channel | null;
+  private messageIndex = 0;
+  private clients: ReturnType<typeof clientsFactory>;
+  private _socket: Socket;
 
-let channel: Channel | null;
-let messageIndex = 0;
-let clients: ReturnType<typeof clientsFactory>;
-let _socket: Socket;
-let provideJwtToken: () => string;
+  private provideJwtToken: () => string;
+  private getConnectionsCount: () => number;
 
-export default new (class Live {
   initialize(options: Options) {
-    const live = this;
-
-    clients = clientsFactory(
+    this.clients = clientsFactory(
       (moduleShortid, revision, operation) => {
-        live.send('operation', {
-          moduleShortid,
-          operation,
-          revision,
-        });
+        this.sendOperation(moduleShortid, revision, operation);
       },
       (moduleShortid, operation) => {
         options.onApplyOperation({
@@ -58,48 +54,49 @@ export default new (class Live {
         });
       }
     );
-    provideJwtToken = options.provideJwtToken;
+    this.provideJwtToken = options.provideJwtToken;
+    this.getConnectionsCount = options.getConnectionsCount;
   }
 
   getSocket() {
-    return _socket || this.connect();
+    return this._socket || this.connect();
   }
 
   connect(): Socket {
-    if (!_socket) {
+    if (!this._socket) {
       const protocol = process.env.LOCAL_SERVER ? 'ws' : 'wss';
-      _socket = new Socket(`${protocol}://${location.host}/socket`, {
+      this._socket = new Socket(`${protocol}://${location.host}/socket`, {
         params: {
-          guardian_token: provideJwtToken(),
+          guardian_token: this.provideJwtToken(),
         },
       });
 
-      _socket.connect();
-      window.socket = _socket;
-      debug('Connecting to socket', _socket);
+      this._socket.connect();
+      window.socket = this._socket;
+      this.debug('Connecting to socket', this._socket);
     }
 
-    return _socket;
+    return this._socket;
   }
 
   disconnect() {
     return new Promise((resolve, reject) => {
-      if (!channel) {
+      if (!this.channel) {
         resolve({});
         return;
       }
 
-      channel
+      this.channel
         .leave()
         .receive('ok', resp => {
-          if (!channel) {
+          if (!this.channel) {
             return resolve({});
           }
 
-          channel.onMessage = d => d;
-          channel = null;
-          sentMessages.clear();
-          messageIndex = 0;
+          this.channel.onMessage = d => d;
+          this.channel = null;
+          this.sentMessages.clear();
+          this.messageIndex = 0;
 
           return resolve(resp);
         })
@@ -110,9 +107,9 @@ export default new (class Live {
 
   joinChannel(roomId: string): Promise<JoinChannelResponse> {
     return new Promise((resolve, reject) => {
-      channel = this.getSocket().channel(`live:${roomId}`, { version: 2 });
+      this.channel = this.getSocket().channel(`live:${roomId}`, { version: 2 });
 
-      channel
+      this.channel
         .join()
         .receive('ok', resp => {
           const result = camelizeKeys(resp) as JoinChannelResponse;
@@ -130,18 +127,18 @@ export default new (class Live {
       data: object;
     }) => {}
   ) {
-    if (!channel) {
+    if (!this.channel) {
       return;
     }
 
-    channel.onMessage = (event: any, data: any) => {
+    this.channel.onMessage = (event: any, data: any) => {
       const disconnected =
         (data == null || Object.keys(data).length === 0) &&
         event === 'phx_error';
       const alteredEvent = disconnected ? 'connection-loss' : event;
 
       const _isOwnMessage = Boolean(
-        data && data._messageId && sentMessages.delete(data._messageId)
+        data && data._messageId && this.sentMessages.delete(data._messageId)
       );
 
       action({
@@ -155,14 +152,18 @@ export default new (class Live {
   }
 
   send(event: string, payload: { _messageId?: string; [key: string]: any }) {
-    const _messageId = identifier + messageIndex++;
+    if (this.getConnectionsCount() < 2) {
+      return Promise.resolve();
+    }
+
+    const _messageId = this.identifier + this.messageIndex++;
     // eslint-disable-next-line
     payload._messageId = _messageId;
-    sentMessages.set(_messageId, payload);
+    this.sentMessages.set(_messageId, payload);
 
     return new Promise((resolve, reject) => {
-      if (channel) {
-        channel
+      if (this.channel) {
+        this.channel
           .push(event, payload)
           .receive('ok', resolve)
           .receive('error', reject);
@@ -202,7 +203,7 @@ export default new (class Live {
     }
 
     try {
-      clients.get(moduleShortid).applyClient(operation);
+      this.clients.get(moduleShortid).applyClient(operation);
     } catch (e) {
       // Something went wrong, probably a sync mismatch. Request new version
       this.send('live:module_state', {});
@@ -309,31 +310,59 @@ export default new (class Live {
     });
   }
 
+  getClient(moduleShortid: string) {
+    return this.clients.get(moduleShortid);
+  }
+
   getAllClients() {
-    return clients.getAll();
+    return this.clients.getAll();
   }
 
   applyClient(moduleShortid: string, operation: any) {
-    return clients
+    return this.clients
       .get(moduleShortid)
       .applyClient(TextOperation.fromJSON(operation));
   }
 
   applyServer(moduleShortid: string, operation: any) {
-    return clients
+    return this.clients
       .get(moduleShortid)
       .applyServer(TextOperation.fromJSON(operation));
   }
 
   serverAck(moduleShortid: string) {
-    return clients.get(moduleShortid).serverAck();
+    return this.clients.get(moduleShortid).serverAck();
   }
 
   createClient(moduleShortid: string, revision: number) {
-    return clients.create(moduleShortid, revision);
+    return this.clients.create(moduleShortid, revision);
   }
 
   resetClients() {
-    clients.clear();
+    this.clients.clear();
+  }
+
+  private operationToElixir(ot) {
+    return ot.map(op => {
+      if (typeof op === 'number') {
+        if (op < 0) {
+          return { d: -op };
+        }
+
+        return op;
+      }
+
+      return { i: op };
+    });
   }
-})();
+
+  private sendOperation(moduleShortid, revision, operation) {
+    this.send('operation', {
+      moduleShortid,
+      operation: this.operationToElixir(operation.toJSON()),
+      revision: Number(revision),
+    });
+  }
+}
+
+export default new Live();

packages/app/src/app/overmind/effects/live/ot.d.ts

@@ -0,0 +1,16 @@
+// eslint-disable-next-line
+declare module 'ot' {
+  export class Client {
+    constructor(revision: number);
+    revision: number;
+    serverAck(): void;
+    clientAck(): void;
+    applyClient(operation: TextOperation): void;
+    applyServer(operation: TextOperation): void;
+    serverReconnect(): void;
+  }
+  export class TextOperation {
+    static fromJSON(operation: object): TextOperation;
+    compose(operation: TextOperation): TextOperation;
+  }
+}

packages/app/src/app/overmind/namespaces/editor/actions.ts

@@ -9,6 +9,8 @@ import {
 } from '@codesandbox/common/lib/types';
 import { getTextOperation } from '@codesandbox/common/lib/utils/diff';
 import { convertTypeToStatus } from '@codesandbox/common/lib/utils/notifications';
+import { hasPermission } from '@codesandbox/common/lib/utils/permission';
+import { NotificationStatus } from '@codesandbox/notifications';
 import { Action, AsyncAction } from 'app/overmind';
 import { withLoadApp, withOwnedSandbox } from 'app/overmind/factories';
 import {
@@ -19,11 +21,9 @@ import {
 import { clearCorrectionsFromAction } from 'app/utils/corrections';
 import { json } from 'overmind';
 
-import { hasPermission } from '@codesandbox/common/lib/utils/permission';
-import { NotificationStatus } from '@codesandbox/notifications';
 import eventToTransform from '../../utils/event-to-transform';
-import * as internalActions from './internalActions';
 import { SERVER } from '../../utils/items';
+import * as internalActions from './internalActions';
 
 export const internal = internalActions;
 
@@ -188,7 +188,11 @@ export const codeSaved: AsyncAction<{
   moduleShortid: string;
   cbID: string | null;
 }> = withOwnedSandbox(
-  async ({ actions }, { code, moduleShortid, cbID }) => {
+  async ({ actions, effects }, { code, moduleShortid, cbID }) => {
+    const pendingOperation = effects.live.getClient(moduleShortid).flush();
+
+    // eslint-disable-next-line
+    console.log('OKAY, READY TO SAVE MODULE CODE OPERATION', pendingOperation);
     actions.editor.internal.saveCode({
       code,
       moduleShortid,