Mutation pump (ampproject#320)

* Mutation pump

* More documentation on the pump semantics

* Rename TransferPhase to Phase and provide tests

* missing files

* removed browser-env, use jsdom

* expand size

config/tsconfig.tests.main-thread.json

@@ -0,0 +1,6 @@
+{
+  "extends": "../src/test/main-thread/tsconfig.json",
+  "compilerOptions": {
+    "module": "commonjs"
+  }
+}

package.json

@@ -17,6 +17,7 @@
     "tsc:test:worker": "tsc -p config/tsconfig.test.worker-thread.json",
     "tsc:test:main": "tsc -p config/tsconfig.test.main-thread.json",
     "tsc:test:tests": "tsc -p config/tsconfig.test.json",
+    "tsc:test:tests-main": "tsc -p config/tsconfig.tests.main-thread.json",
     "tsc:build:worker": "tsc -p config/tsconfig.build.worker-thread.json",
     "tsc:build:main": "tsc -p config/tsconfig.build.main-thread.json",
     "clean": "rimraf output dist",
@@ -55,6 +56,7 @@
     "cross-var": "1.1.0",
     "esm": "3.2.14",
     "husky": "1.3.1",
+    "jsdom": "^13.2.0",
     "lint-staged": "8.1.5",
     "magic-string": "0.25.2",
     "np": "4.0.2",
@@ -96,7 +98,7 @@
     {
       "path": "./dist/index.mjs",
       "compression": "brotli",
-      "maxSize": "2.8 kB"
+      "maxSize": "2.82 kB"
     },
     {
       "path": "./dist/index.safe.mjs",

src/main-thread/callbacks.ts

@@ -15,6 +15,14 @@
  */
 
 import { MessageFromWorker, MessageToWorker } from '../transfer/Messages';
+import { Phase } from '../transfer/phase';
+
+/**
+ * The callback for `onMutationPump`. If specified, this callback will be called
+ * for the new set of mutations pending. The callback can either immediately
+ * call `flush()`, or it can reject mutations, or it can batch them further.
+ */
+export type MutationPumpFunction = (flush: Function, phase: Phase) => void;
 
 export interface WorkerCallbacks {
   // Called when worker consumes the page's initial DOM state.
@@ -25,4 +33,6 @@ export interface WorkerCallbacks {
   onSendMessage?: (message: MessageToWorker) => void;
   // Called after a message is received from the worker.
   onReceiveMessage?: (message: MessageFromWorker) => void;
+  // Called to schedule mutation phase. See `MutationPumpFunction`.
+  onMutationPump?: MutationPumpFunction;
 }

src/main-thread/install.ts

@@ -18,6 +18,7 @@ import { DebuggingContext } from './debugging';
 import { MutationFromWorker, MessageType, MessageFromWorker } from '../transfer/Messages';
 import { MutatorProcessor } from './mutator';
 import { NodeContext } from './nodes';
+import { Phase } from '../transfer/phase';
 import { Strings } from './strings';
 import { TransferrableKeys } from '../transfer/TransferrableKeys';
 import { WorkerCallbacks } from './callbacks';
@@ -74,23 +75,23 @@ export function install(
     if (workerDOMScript && authorScript && authorScriptURL) {
       const workerContext = new WorkerContext(baseElement, workerDOMScript, authorScript, authorScriptURL, callbacks);
       const worker = workerContext.getWorker();
-      const mutatorContext = new MutatorProcessor(strings, nodeContext, workerContext, sanitizer);
+      const mutatorContext = new MutatorProcessor(strings, nodeContext, workerContext, callbacks && callbacks.onMutationPump, sanitizer);
       worker.onmessage = (message: MessageFromWorker) => {
         const { data } = message;
         const type = data[TransferrableKeys.type];
         if (!ALLOWABLE_MESSAGE_TYPES.includes(type)) {
           return;
         }
-        // TODO(KB): Hydration has special rules limiting the types of allowed mutations.
-        // Re-introduce Hydration and add a specialized handler.
+        const phase = type == MessageType.HYDRATE ? Phase.Hydrating : Phase.Mutating;
         mutatorContext.mutate(
+          phase,
           (data as MutationFromWorker)[TransferrableKeys.nodes],
           (data as MutationFromWorker)[TransferrableKeys.strings],
           (data as MutationFromWorker)[TransferrableKeys.mutations],
         );
         // Invoke callbacks after hydrate/mutate processing so strings etc. are stored.
         if (callbacks) {
-          if (type === MessageType.HYDRATE && callbacks.onHydration) {
+          if (phase === Phase.Hydrating && callbacks.onHydration) {
             callbacks.onHydration();
           }
           if (callbacks.onReceiveMessage) {
@@ -130,5 +131,7 @@ function wrapCallbacks(debuggingContext: DebuggingContext, callbacks?: WorkerCal
         callbacks.onReceiveMessage(readable as any);
       }
     },
+    // Passthrough callbacks:
+    onMutationPump: callbacks && callbacks.onMutationPump,
   };
 }

src/main-thread/main-thread.d.ts

@@ -20,6 +20,7 @@ declare interface DOMPurify {
   removeAllHooks(): void;
   sanitize(dirty: string | Node, cfg: Object): string | Node;
 }
+
 declare module 'dompurify' {
   var purify: DOMPurify;
   export default purify;

src/main-thread/mutator.ts

@@ -17,18 +17,24 @@
 import { TransferrableMutationRecord } from '../transfer/TransferrableRecord';
 import { TransferrableKeys } from '../transfer/TransferrableKeys';
 import { MutationRecordType } from '../worker-thread/MutationRecord';
+import { Phase } from '../transfer/phase';
 import { TransferrableNode } from '../transfer/TransferrableNodes';
 import { NodeContext } from './nodes';
 import { Strings } from './strings';
 import { WorkerContext } from './worker';
 import { EventSubscriptionProcessor } from './commands/event-subscription';
 import { BoundingClientRectProcessor } from './commands/bounding-client-rect';
+import { MutationPumpFunction } from './callbacks';
 
 export class MutatorProcessor {
   private strings: Strings;
   private nodeContext: NodeContext;
+  private mutationPump: MutationPumpFunction;
+  private boundSyncFlush: () => void;
   private mutationQueue: Array<TransferrableMutationRecord>;
-  private pendingMutations: boolean;
+  private stringQueue: Array<string>;
+  private nodeQueue: Array<TransferrableNode>;
+  private pendingQueue: boolean;
   private sanitizer: Sanitizer | undefined;
   private mutators: {
     [key: number]: (mutation: TransferrableMutationRecord, target: Node) => void;
@@ -38,14 +44,18 @@ export class MutatorProcessor {
    * @param strings
    * @param nodeContext
    * @param workerContext
-   * @param passedSanitizer Sanitizer to apply to content if needed.
+   * @param sanitizer Sanitizer to apply to content if needed.
    */
-  constructor(strings: Strings, nodeContext: NodeContext, workerContext: WorkerContext, passedSanitizer?: Sanitizer) {
+  constructor(strings: Strings, nodeContext: NodeContext, workerContext: WorkerContext, mutationPump?: MutationPumpFunction, sanitizer?: Sanitizer) {
     this.strings = strings;
     this.nodeContext = nodeContext;
-    this.sanitizer = passedSanitizer;
+    this.sanitizer = sanitizer;
+    this.mutationPump = mutationPump || requestAnimationFrame.bind(null);
+    this.boundSyncFlush = this.syncFlush.bind(this);
+    this.stringQueue = [];
+    this.nodeQueue = [];
     this.mutationQueue = [];
-    this.pendingMutations = false;
+    this.pendingQueue = false;
 
     const eventSubscriptionProcessor = new EventSubscriptionProcessor(strings, nodeContext, workerContext);
     const boundingClientRectProcessor = new BoundingClientRectProcessor(nodeContext, workerContext);
@@ -62,23 +72,18 @@ export class MutatorProcessor {
 
   /**
    * Process MutationRecords from worker thread applying changes to the existing DOM.
+   * @param phase
    * @param nodes New nodes to add in the main thread with the incoming mutations.
    * @param stringValues Additional string values to use in decoding messages.
    * @param mutations Changes to apply in both graph shape and content of Elements.
    */
-  mutate(nodes: Array<TransferrableNode>, stringValues: Array<string>, mutations: Array<TransferrableMutationRecord>): void {
-    //mutations: TransferrableMutationRecord[]): void {
-    // TODO(KB): Restore signature requiring lastMutationTime. (lastGestureTime: number, mutations: TransferrableMutationRecord[])
-    // if (performance.now() || Date.now() - lastGestureTime > GESTURE_TO_MUTATION_THRESHOLD) {
-    //   return;
-    // }
-    // this.lastGestureTime = lastGestureTime;
-    this.strings.storeValues(stringValues);
-    nodes.forEach(node => this.nodeContext.createNode(node, this.sanitizer));
+  mutate(phase: Phase, nodes: Array<TransferrableNode>, stringValues: Array<string>, mutations: Array<TransferrableMutationRecord>): void {
+    this.stringQueue = this.stringQueue.concat(stringValues);
+    this.nodeQueue = this.nodeQueue.concat(nodes);
     this.mutationQueue = this.mutationQueue.concat(mutations);
-    if (!this.pendingMutations) {
-      this.pendingMutations = true;
-      requestAnimationFrame(this.syncFlush.bind(this));
+    if (!this.pendingQueue) {
+      this.pendingQueue = true;
+      this.mutationPump(this.boundSyncFlush, phase);
     }
   }
 
@@ -89,6 +94,14 @@ export class MutatorProcessor {
    * Investigations in using asyncFlush to resolve are worth considering.
    */
   private syncFlush(): void {
+    this.pendingQueue = false;
+
+    this.strings.storeValues(this.stringQueue);
+    this.stringQueue.length = 0;
+
+    this.nodeQueue.forEach(node => this.nodeContext.createNode(node, this.sanitizer));
+    this.nodeQueue.length = 0;
+
     this.mutationQueue.forEach(mutation => {
       const nodeId = mutation[TransferrableKeys.target];
       const node = this.nodeContext.getNode(nodeId);
@@ -98,8 +111,7 @@ export class MutatorProcessor {
       }
       this.mutators[mutation[TransferrableKeys.type]](mutation, node);
     });
-    this.mutationQueue = [];
-    this.pendingMutations = false;
+    this.mutationQueue.length = 0;
   }
 
   private mutateChildList(mutation: TransferrableMutationRecord, target: HTMLElement) {

src/main-thread/tsconfig.json

@@ -3,7 +3,7 @@
   "compilerOptions": {
     "lib": [
       "es2017",
-      "dom",
+      "dom"
     ],
     "outDir": "../../output",
     "sourceMap": true,
@@ -14,9 +14,9 @@
     "strictNullChecks": true,
     "noImplicitAny": true,
     "noImplicitThis": true,
-    "alwaysStrict": true,
+    "alwaysStrict": true
   },
   "include": [
-    "*.ts"
+    "**/*.ts"
   ]
-}
+}