Implement cursor offsets rather than intervals

surminus · surminus · commit 0f71ac9ffc39 · 2023-08-10T15:24:05.000+01:00
Instead of calculating an even interval based upon the length of the
buffer and the batchtime, instead pass along movement offsets as part of
the message, and replay these offsets as part of the dispensing.

This ensures you get a more accurate representation of cursor movements.

The first offset is also 0, which removes the slight delay that was
present in the original implementation, since the calculated interval
was always the same.

Since we are immediately invoking the all the data in the buffer to
create the timeouts, we can simplify the code to always just use
whatever data is present in the buffer as the gate to whether we should
call `emitFromBatch()`. This means we can remove the logic related to
`handlerRunning`.

Due to the way we are immediately processing the buffer, it means we
cannot test in the same way we did previously, so tests have been
amended to accomodate this.
diff --git a/src/CursorBatching.ts b/src/CursorBatching.ts
@@ -4,10 +4,10 @@ import { CURSOR_UPDATE } from './CursorConstants.js';
 import type { CursorUpdate } from './types.js';
 import type { CursorsOptions } from './types.js';
 
-type OutgoingBuffer = Pick<CursorUpdate, 'position' | 'data'>[];
+type OutgoingBuffer = { cursor: Pick<CursorUpdate, 'position' | 'data'>; offset: number };
 
 export default class CursorBatching {
-  outgoingBuffers: OutgoingBuffer = [];
+  outgoingBuffers: OutgoingBuffer[] = [];
 
   batchTime: number;
 
@@ -20,15 +20,31 @@ export default class CursorBatching {
   // Set to `true` if there is more than one user listening to cursors
   shouldSend: boolean = false;
 
+  // Used for tracking offsets in the buffer
+  bufferStartTimestamp: number = 0;
+
   constructor(readonly outboundBatchInterval: CursorsOptions['outboundBatchInterval']) {
     this.batchTime = outboundBatchInterval;
   }
 
   pushCursorPosition(channel: Types.RealtimeChannelPromise, cursor: Pick<CursorUpdate, 'position' | 'data'>) {
     // Ignore the cursor update if there is no one listening
     if (!this.shouldSend) return;
+
+    const timestamp = new Date().getTime();
+
+    let offset: number;
+    // First update in the buffer is always 0
+    if (this.outgoingBuffers.length === 0) {
+      offset = 0;
+      this.bufferStartTimestamp = timestamp;
+    } else {
+      // Add the offset compared to the first update in the buffer
+      offset = timestamp - this.bufferStartTimestamp;
+    }
+
     this.hasMovement = true;
-    this.pushToBuffer(cursor);
+    this.pushToBuffer({ cursor, offset });
     this.publishFromBuffer(channel, CURSOR_UPDATE);
   }
 
@@ -40,7 +56,7 @@ export default class CursorBatching {
     this.batchTime = batchTime;
   }
 
-  private pushToBuffer(value: Pick<CursorUpdate, 'position' | 'data'>) {
+  private pushToBuffer(value: OutgoingBuffer) {
     this.outgoingBuffers.push(value);
   }
 
diff --git a/src/CursorDispensing.ts b/src/CursorDispensing.ts
@@ -1,47 +1,26 @@
-import { clamp } from './utilities/math.js';
-
 import { type CursorUpdate } from './types.js';
 import { type RealtimeMessage } from './utilities/types.js';
 
 export default class CursorDispensing {
-  private buffer: Record<string, CursorUpdate[]> = {};
-  private handlerRunning: boolean = false;
-  private timerIds: ReturnType<typeof setTimeout>[] = [];
-
-  constructor(private emitCursorUpdate: (update: CursorUpdate) => void, private getCurrentBatchTime: () => number) {}
-
-  emitFromBatch(batchDispenseInterval: number) {
-    if (!this.bufferHaveData()) {
-      this.handlerRunning = false;
-      return;
-    }
-
-    this.handlerRunning = true;
+  private buffer: Record<string, { cursor: CursorUpdate; offset: number }[]> = {};
 
-    const processBuffer = () => {
-      for (let connectionId in this.buffer) {
-        const buffer = this.buffer[connectionId];
-        const update = buffer.shift();
+  constructor(private emitCursorUpdate: (update: CursorUpdate) => void) {}
 
-        if (!update) continue;
-        this.emitCursorUpdate(update);
-      }
+  setEmitCursorUpdate(update: CursorUpdate) {
+    this.emitCursorUpdate(update);
+  }
 
-      if (this.bufferHaveData()) {
-        this.emitFromBatch(this.calculateDispenseInterval());
-      } else {
-        this.handlerRunning = false;
-      }
+  emitFromBatch() {
+    for (let connectionId in this.buffer) {
+      const buffer = this.buffer[connectionId];
+      const update = buffer.shift();
 
-      this.timerIds.shift();
-    };
+      if (!update) continue;
+      setTimeout(() => this.setEmitCursorUpdate(update.cursor), update.offset);
+    }
 
-    if (typeof document !== 'undefined' && document.visibilityState === 'hidden') {
-      this.timerIds.forEach((id) => clearTimeout(id));
-      this.timerIds = [];
-      processBuffer();
-    } else {
-      this.timerIds.push(setTimeout(processBuffer, batchDispenseInterval));
+    if (this.bufferHaveData()) {
+      this.emitFromBatch();
     }
   }
 
@@ -53,33 +32,29 @@ export default class CursorDispensing {
     );
   }
 
-  calculateDispenseInterval(): number {
-    const bufferLengths = Object.entries(this.buffer).map(([, v]) => v.length);
-    const highest = bufferLengths.sort()[bufferLengths.length - 1];
-    const finalOutboundBatchInterval = this.getCurrentBatchTime();
-    return Math.floor(clamp(finalOutboundBatchInterval / highest, 1, 1000 / 15));
-  }
-
   processBatch(message: RealtimeMessage) {
-    const updates: CursorUpdate[] = message.data || [];
+    const updates: { cursor: CursorUpdate; offset: number }[] = message.data || [];
 
-    updates.forEach((update) => {
+    updates.forEach((update: { cursor: CursorUpdate; offset: number }) => {
       const enhancedMsg = {
-        clientId: message.clientId,
-        connectionId: message.connectionId,
-        position: update.position,
-        data: update.data,
+        cursor: {
+          clientId: message.clientId,
+          connectionId: message.connectionId,
+          position: update.cursor.position,
+          data: update.cursor.data,
+        },
+        offset: update.offset,
       };
 
-      if (this.buffer[enhancedMsg.connectionId]) {
-        this.buffer[enhancedMsg.connectionId].push(enhancedMsg);
+      if (this.buffer[enhancedMsg.cursor.connectionId]) {
+        this.buffer[enhancedMsg.cursor.connectionId].push(enhancedMsg);
       } else {
-        this.buffer[enhancedMsg.connectionId] = [enhancedMsg];
+        this.buffer[enhancedMsg.cursor.connectionId] = [enhancedMsg];
       }
     });
 
-    if (!this.handlerRunning && this.bufferHaveData()) {
-      this.emitFromBatch(this.calculateDispenseInterval());
+    if (this.bufferHaveData()) {
+      this.emitFromBatch();
     }
   }
 }
diff --git a/src/Cursors.test.ts b/src/Cursors.test.ts
@@ -67,7 +67,10 @@ describe('Cursors', () => {
     it<CursorsTestContext>('emits a cursorsUpdate event', ({ space, dispensing, batching, fakeMessageStub }) => {
       const fakeMessage = {
         ...fakeMessageStub,
-        data: [{ position: { x: 1, y: 1 } }, { position: { x: 1, y: 2 }, data: { color: 'red' } }],
+        data: [
+          { cursor: { position: { x: 1, y: 1 } } },
+          { cursor: { position: { x: 1, y: 2 }, data: { color: 'red' } } },
+        ],
       };
 
       const spy = vitest.fn();
@@ -151,16 +154,19 @@ describe('Cursors', () => {
 
       it<CursorsTestContext>('creates an outgoingBuffer for a new cursor movement', ({ batching, channel }) => {
         batching.pushCursorPosition(channel, { position: { x: 1, y: 1 }, data: {} });
-        expect(batching.outgoingBuffers).toEqual([{ position: { x: 1, y: 1 }, data: {} }]);
+        expect(batching.outgoingBuffers).toEqual([{ cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 }]);
       });
 
       it<CursorsTestContext>('adds cursor data to an existing buffer', ({ batching, channel }) => {
+        vi.useFakeTimers();
         batching.pushCursorPosition(channel, { position: { x: 1, y: 1 }, data: {} });
-        expect(batching.outgoingBuffers).toEqual([{ position: { x: 1, y: 1 }, data: {} }]);
+        expect(batching.outgoingBuffers).toEqual([{ cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 }]);
+
+        vi.advanceTimersByTime(10);
         batching.pushCursorPosition(channel, { position: { x: 2, y: 2 }, data: {} });
         expect(batching.outgoingBuffers).toEqual([
-          { position: { x: 1, y: 1 }, data: {} },
-          { position: { x: 2, y: 2 }, data: {} },
+          { cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 },
+          { cursor: { position: { x: 2, y: 2 }, data: {} }, offset: 10 },
         ]);
       });
 
@@ -195,15 +201,17 @@ describe('Cursors', () => {
 
       it<CursorsTestContext>('should publish the cursor buffer', async ({ batching, channel }) => {
         batching.hasMovement = true;
-        batching.outgoingBuffers = [{ position: { x: 1, y: 1 }, data: {} }];
+        batching.outgoingBuffers = [{ cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 }];
         const spy = vi.spyOn(channel, 'publish');
         await batching['batchToChannel'](channel, CURSOR_UPDATE);
-        expect(spy).toHaveBeenCalledWith(CURSOR_UPDATE, [{ position: { x: 1, y: 1 }, data: {} }]);
+        expect(spy).toHaveBeenCalledWith(CURSOR_UPDATE, [
+          { cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 },
+        ]);
       });
 
       it<CursorsTestContext>('should clear the buffer', async ({ batching, channel }) => {
         batching.hasMovement = true;
-        batching.outgoingBuffers = [{ position: { x: 1, y: 1 }, data: {} }];
+        batching.outgoingBuffers = [{ cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 }];
         await batching['batchToChannel'](channel, CURSOR_UPDATE);
         expect(batching.outgoingBuffers).toEqual([]);
       });
@@ -233,28 +241,12 @@ describe('Cursors', () => {
         expect(spy).not.toHaveBeenCalled();
       });
 
-      it<CursorsTestContext>('does not call emitFromBatch if the loop is already running', async ({
-        dispensing,
-        fakeMessageStub,
-      }) => {
-        const spy = vi.spyOn(dispensing, 'emitFromBatch');
-
-        const fakeMessage = {
-          ...fakeMessageStub,
-          data: [{ position: { x: 1, y: 1 } }],
-        };
-
-        dispensing['handlerRunning'] = true;
-        dispensing.processBatch(fakeMessage);
-        expect(spy).not.toHaveBeenCalled();
-      });
-
       it<CursorsTestContext>('call emitFromBatch if there are updates', async ({ dispensing, fakeMessageStub }) => {
         const spy = vi.spyOn(dispensing, 'emitFromBatch');
 
         const fakeMessage = {
           ...fakeMessageStub,
-          data: [{ position: { x: 1, y: 1 } }],
+          data: [{ cursor: { position: { x: 1, y: 1 } } }],
         };
 
         dispensing.processBatch(fakeMessage);
@@ -268,70 +260,70 @@ describe('Cursors', () => {
         const fakeMessage = {
           ...fakeMessageStub,
           data: [
-            { position: { x: 1, y: 1 } },
-            { position: { x: 2, y: 3 }, data: { color: 'blue' } },
-            { position: { x: 5, y: 4 } },
+            { cursor: { position: { x: 1, y: 1 } }, offset: 10 },
+            { cursor: { position: { x: 2, y: 3 }, data: { color: 'blue' } }, offset: 20 },
+            { cursor: { position: { x: 5, y: 4 } }, offset: 30 },
           ],
         };
+        vi.useFakeTimers();
 
+        const spy = vi.spyOn(dispensing, 'setEmitCursorUpdate');
         dispensing.processBatch(fakeMessage);
-        expect(dispensing['buffer']).toEqual({
-          connectionId: [
-            {
-              position: { x: 1, y: 1 },
-              data: undefined,
-              clientId: 'clientId',
-              connectionId: 'connectionId',
-            },
-            {
-              position: { x: 2, y: 3 },
-              data: { color: 'blue' },
-              clientId: 'clientId',
-              connectionId: 'connectionId',
-            },
-            {
-              position: { x: 5, y: 4 },
-              data: undefined,
-              clientId: 'clientId',
-              connectionId: 'connectionId',
-            },
-          ],
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledWith({
+          position: { x: 1, y: 1 },
+          data: undefined,
+          clientId: 'clientId',
+          connectionId: 'connectionId',
+        });
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledWith({
+          position: { x: 2, y: 3 },
+          data: { color: 'blue' },
+          clientId: 'clientId',
+          connectionId: 'connectionId',
         });
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledWith({
+          position: { x: 5, y: 4 },
+          data: undefined,
+          clientId: 'clientId',
+          connectionId: 'connectionId',
+        });
+        expect(spy).toHaveBeenCalledTimes(3);
       });
 
-      it<CursorsTestContext>('runs until the batch is empty', async ({ dispensing, batching, fakeMessageStub }) => {
+      it<CursorsTestContext>('runs until the batch is empty', async ({ dispensing, fakeMessageStub }) => {
         vi.useFakeTimers();
 
         const fakeMessage = {
           ...fakeMessageStub,
           data: [
-            { position: { x: 1, y: 1 } },
-            { position: { x: 2, y: 3 }, data: { color: 'blue' } },
-            { position: { x: 5, y: 4 } },
+            { cursor: { position: { x: 1, y: 1 } }, offset: 10 },
+            { cursor: { position: { x: 2, y: 3 }, data: { color: 'blue' } }, offset: 20 },
+            { cursor: { position: { x: 5, y: 4 } }, offset: 30 },
           ],
         };
 
-        expect(dispensing['handlerRunning']).toBe(false);
+        const spy = vi.spyOn(dispensing, 'setEmitCursorUpdate');
+
         expect(dispensing.bufferHaveData()).toBe(false);
 
         dispensing.processBatch(fakeMessage);
-        expect(dispensing['buffer']['connectionId']).toHaveLength(3);
-        expect(dispensing['handlerRunning']).toBe(true);
-        expect(dispensing.bufferHaveData()).toBe(true);
-        vi.advanceTimersByTime(batching.batchTime / 2);
-
-        expect(dispensing['buffer']['connectionId']).toHaveLength(2);
-        expect(dispensing['handlerRunning']).toBe(true);
-        expect(dispensing.bufferHaveData()).toBe(true);
-
-        vi.advanceTimersByTime(batching.batchTime / 2);
-        expect(dispensing['buffer']['connectionId']).toHaveLength(1);
-        expect(dispensing['handlerRunning']).toBe(true);
-        expect(dispensing.bufferHaveData()).toBe(true);
-
-        vi.advanceTimersByTime(batching.batchTime);
-        expect(dispensing['buffer']['connectionId']).toHaveLength(0);
-        expect(dispensing['handlerRunning']).toBe(false);
+        expect(spy).toHaveBeenCalledTimes(0);
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledTimes(1);
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledTimes(2);
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledTimes(3);
+
         expect(dispensing.bufferHaveData()).toBe(false);
 
         vi.useRealTimers();
@@ -498,6 +490,7 @@ describe('Cursors', () => {
             x: 2,
             y: 3,
           },
+          offset: 0,
         },
       };
 
diff --git a/src/Cursors.ts b/src/Cursors.ts
@@ -35,8 +35,7 @@ export default class Cursors extends EventEmitter<CursorsEventMap> {
     this.cursorBatching = new CursorBatching(this.options.outboundBatchInterval);
 
     const emitCursorUpdate = (update: CursorUpdate): void => this.emit('cursorsUpdate', update);
-    const getCurrentBatchTime: () => number = () => this.cursorBatching.batchTime;
-    this.cursorDispensing = new CursorDispensing(emitCursorUpdate, getCurrentBatchTime);
+    this.cursorDispensing = new CursorDispensing(emitCursorUpdate);
   }
 
   /**

Original file line number	Diff line number	Diff line change
`@@ -35,8 +35,7 @@ export default class Cursors extends EventEmitter<CursorsEventMap> {`
`35`	`35`	`this.cursorBatching = new CursorBatching(this.options.outboundBatchInterval);`
`36`	`36`
`37`	`37`	`const emitCursorUpdate = (update: CursorUpdate): void => this.emit('cursorsUpdate', update);`
`38`		`- const getCurrentBatchTime: () => number = () => this.cursorBatching.batchTime;`
`39`		`- this.cursorDispensing = new CursorDispensing(emitCursorUpdate, getCurrentBatchTime);`
	`38`	`+ this.cursorDispensing = new CursorDispensing(emitCursorUpdate);`
`40`	`39`	`}`
`41`	`40`
`42`	`41`	`/**`