feat: enhance AtemSocket to handle caching

src/lib/__tests__/atemSocket.spec.ts

@@ -3,6 +3,7 @@
 import {
 	CutCommand,
 	ProductIdentifierCommand,
+	TimeCommand,
 	VersionCommand,
 	ProgramInputUpdateCommand,
 	PreviewInputUpdateCommand,
@@ -11,7 +12,7 @@ import {
 	DeserializedCommand,
 } from '../../commands'
 import { ProtocolVersion, Model } from '../../enums'
-import { AtemSocket } from '../atemSocket'
+import { AtemSocket, ThreadedPayload } from '../atemSocket'
 import { ThreadedClass, ThreadedClassManager } from 'threadedclass'
 import { Buffer } from 'buffer'
 import { CommandParser } from '../atemCommandParser'
@@ -24,7 +25,7 @@ jest.mock('../atemSocketChild')
 class AtemSocketChildMock implements AtemSocketChild {
 	public onDisconnect: () => Promise<void>
 	public onLog: (message: string) => Promise<void>
-	public onCommandsReceived: (payload: Buffer, packetId: number) => Promise<void>
+	public onCommandsReceived: (payload: ThreadedPayload, packetId: number) => Promise<void>
 	public onPacketsAcknowledged: (ids: Array<{ packetId: number; trackingId: number }>) => Promise<void>
 
 	constructor() {
@@ -49,7 +50,7 @@ const AtemSocketChildSingleton = new AtemSocketChildMock()
 		_opts: any,
 		onDisconnect: () => Promise<void>,
 		onLog: (message: string) => Promise<void>,
-		onCommandsReceived: (payload: Buffer, packetId: number) => Promise<void>,
+		onCommandsReceived: (payload: ThreadedPayload, packetId: number) => Promise<void>,
 		onPacketsAcknowledged: (ids: Array<{ packetId: number; trackingId: number }>) => Promise<void>
 	) => {
 		AtemSocketChildSingleton.onDisconnect = onDisconnect
@@ -467,6 +468,144 @@ describe('AtemSocket', () => {
 		// A change with the command
 		expect(change).toHaveBeenCalledWith([expectedCmd1, expectedCmd2])
 	})
+	test('receive - time command remainder dedupe and cache reset', async () => {
+		const socket = createSocket()
+		expect(getChild(socket)).toBeFalsy()
+
+		await socket.connect()
+		mockClear(true)
+		expect(getChild(socket)).toBeTruthy()
+
+		const error = jest.fn()
+		const change = jest.fn()
+
+		socket.on('error', error)
+		socket.on('receivedCommands', change)
+
+		const parser = (socket as any)._commandParser as CommandParser
+		expect(parser).toBeTruthy()
+		const parserSpy = jest.spyOn(parser, 'commandFromRawName')
+
+		const expectedTime1 = new TimeCommand({ hour: 1, minute: 2, second: 3, frame: 4, dropFrame: false })
+		const expectedTime2 = new TimeCommand({ hour: 1, minute: 2, second: 3, frame: 5, dropFrame: false })
+		const expectedTime3 = new TimeCommand({ hour: 1, minute: 2, second: 3, frame: 6, dropFrame: false })
+		const expectedProgram = new ProgramInputUpdateCommand(0, { source: 0x0123 })
+		const expectedPreview = new PreviewInputUpdateCommand(1, { source: 0x0444 })
+
+		const timeCmd1 = Buffer.concat([
+			Buffer.from([0, 16, 0, 0, ...Buffer.from(TimeCommand.rawName, 'ascii')]),
+			expectedTime1.serialize(),
+		])
+		const timeCmd2 = Buffer.concat([
+			Buffer.from([0, 16, 0, 0, ...Buffer.from(TimeCommand.rawName, 'ascii')]),
+			expectedTime2.serialize(),
+		])
+		const timeCmd3 = Buffer.concat([
+			Buffer.from([0, 16, 0, 0, ...Buffer.from(TimeCommand.rawName, 'ascii')]),
+			expectedTime3.serialize(),
+		])
+		const programCmd = Buffer.from([
+			0,
+			12,
+			0,
+			0,
+			...Buffer.from(ProgramInputUpdateCommand.rawName, 'ascii'),
+			0x00,
+			0x00,
+			0x01,
+			0x23,
+		])
+		const previewCmd = Buffer.from([
+			0,
+			12,
+			0,
+			0,
+			...Buffer.from(PreviewInputUpdateCommand.rawName, 'ascii'),
+			0x01,
+			0x00,
+			0x04,
+			0x44,
+		])
+
+		expect(AtemSocketChildSingleton.onCommandsReceived).toBeDefined()
+		await AtemSocketChildSingleton.onCommandsReceived(Buffer.concat([timeCmd1, previewCmd]), 822)
+		await clock.tickAsync(0)
+		await AtemSocketChildSingleton.onCommandsReceived(Buffer.concat([timeCmd2, previewCmd]), 823)
+		await clock.tickAsync(0)
+		await AtemSocketChildSingleton.onCommandsReceived(Buffer.concat([programCmd, previewCmd]), 824)
+		await clock.tickAsync(0)
+		await AtemSocketChildSingleton.onCommandsReceived(Buffer.concat([timeCmd3, previewCmd]), 825)
+		await clock.tickAsync(0)
+
+		expect(error).toHaveBeenCalledTimes(0)
+		expect(change).toHaveBeenCalledTimes(4)
+		expect(change).toHaveBeenNthCalledWith(1, [expectedTime1, expectedPreview])
+		expect(change).toHaveBeenNthCalledWith(2, [expectedTime2])
+		expect(change).toHaveBeenNthCalledWith(3, [expectedProgram, expectedPreview])
+		expect(change).toHaveBeenNthCalledWith(4, [expectedTime3, expectedPreview])
+
+		expect(parserSpy).toHaveBeenCalledTimes(7)
+		expect(parserSpy).toHaveBeenNthCalledWith(1, TimeCommand.rawName)
+		expect(parserSpy).toHaveBeenNthCalledWith(2, PreviewInputUpdateCommand.rawName)
+		expect(parserSpy).toHaveBeenNthCalledWith(3, TimeCommand.rawName)
+		expect(parserSpy).toHaveBeenNthCalledWith(4, ProgramInputUpdateCommand.rawName)
+		expect(parserSpy).toHaveBeenNthCalledWith(5, PreviewInputUpdateCommand.rawName)
+		expect(parserSpy).toHaveBeenNthCalledWith(6, TimeCommand.rawName)
+		expect(parserSpy).toHaveBeenNthCalledWith(7, PreviewInputUpdateCommand.rawName)
+	})
+	test('receive - payload normalization variants and invalid payload', async () => {
+		const socket = createSocket()
+		expect(getChild(socket)).toBeFalsy()
+
+		await socket.connect()
+		mockClear(true)
+		expect(getChild(socket)).toBeTruthy()
+
+		const error = jest.fn()
+		const change = jest.fn()
+
+		socket.on('error', error)
+		socket.on('receivedCommands', change)
+
+		const parser = (socket as any)._commandParser as CommandParser
+		expect(parser).toBeTruthy()
+		const parserSpy = jest.spyOn(parser, 'commandFromRawName')
+
+		const makeVersionPayload = (a: number, b: number, c: number, d: number): Buffer =>
+			Buffer.from([0, 12, 0, 0, ...Buffer.from(VersionCommand.rawName, 'ascii'), a, b, c, d])
+
+		expect(AtemSocketChildSingleton.onCommandsReceived).toBeDefined()
+
+		const payload1 = makeVersionPayload(0x01, 0x02, 0x03, 0x04)
+		await AtemSocketChildSingleton.onCommandsReceived(new Uint8Array(payload1), 822)
+		await clock.tickAsync(0)
+		expect(error).toHaveBeenCalledTimes(0)
+		expect(change).toHaveBeenCalledWith([new VersionCommand(0x01020304 as ProtocolVersion)])
+		expect(parserSpy).toHaveBeenCalledWith(VersionCommand.rawName)
+
+		error.mockClear()
+		change.mockClear()
+		parserSpy.mockClear()
+
+		const payload2 = makeVersionPayload(0x05, 0x06, 0x07, 0x08)
+		await AtemSocketChildSingleton.onCommandsReceived({ type: 'Buffer', data: Array.from(payload2.values()) }, 823)
+		await clock.tickAsync(0)
+		expect(error).toHaveBeenCalledTimes(0)
+		expect(change).toHaveBeenCalledWith([new VersionCommand(0x05060708 as ProtocolVersion)])
+		expect(parserSpy).toHaveBeenCalledWith(VersionCommand.rawName)
+
+		error.mockClear()
+		change.mockClear()
+		parserSpy.mockClear()
+
+		await AtemSocketChildSingleton.onCommandsReceived('invalid payload' as any, 824)
+		await clock.tickAsync(0)
+
+		expect(error).toHaveBeenCalledTimes(1)
+		expect(error).toHaveBeenCalledWith('Received invalid command payload type: string')
+		expect(change).toHaveBeenCalledTimes(0)
+		expect(parserSpy).toHaveBeenCalledTimes(0)
+	})
 	test('receive - empty buffer', async () => {
 		const socket = createSocket()
 		expect(getChild(socket)).toBeFalsy()

src/lib/atemSocket.ts

@@ -1,7 +1,7 @@
 import { EventEmitter } from 'eventemitter3'
 import { CommandParser } from './atemCommandParser'
 import exitHook = require('exit-hook')
-import { VersionCommand, ISerializableCommand, IDeserializedCommand } from '../commands'
+import { TimeCommand, VersionCommand, ISerializableCommand, IDeserializedCommand } from '../commands'
 import { DEFAULT_PORT } from '../atem'
 import { threadedClass, ThreadedClass, ThreadedClassManager } from 'threadedclass'
 import type { AtemSocketChild, OutboundPacketInfo } from './atemSocketChild'
@@ -25,6 +25,8 @@ export type AtemSocketEvents = {
 	ackPackets: [number[]]
 }
 
+export type ThreadedPayload = Buffer | Uint8Array | { type: 'Buffer'; data: number[] }
+
 export class AtemSocket extends EventEmitter<AtemSocketEvents> {
 	private readonly _debugBuffers: boolean
 	private readonly _disableMultithreaded: boolean
@@ -39,6 +41,7 @@ export class AtemSocket extends EventEmitter<AtemSocketEvents> {
 	private _socketProcess: ThreadedClass<AtemSocketChild> | undefined
 	private _creatingSocket: Promise<void> | undefined
 	private _exitUnsubscribe?: () => void
+	private _lastTimeCommandRemainder: Buffer | undefined
 
 	constructor(options: AtemSocketOptions) {
 		super()
@@ -52,6 +55,7 @@ export class AtemSocket extends EventEmitter<AtemSocketEvents> {
 
 	public async connect(address?: string, port?: number): Promise<void> {
 		this._isDisconnecting = false
+		this._lastTimeCommandRemainder = undefined
 
 		if (address) {
 			this._address = address
@@ -91,6 +95,7 @@ export class AtemSocket extends EventEmitter<AtemSocketEvents> {
 
 	public async disconnect(): Promise<void> {
 		this._isDisconnecting = true
+		this._lastTimeCommandRemainder = undefined
 
 		if (this._socketProcess) {
 			await this._socketProcess.disconnect()
@@ -144,8 +149,14 @@ export class AtemSocket extends EventEmitter<AtemSocketEvents> {
 				async (message: string): Promise<void> => {
 					this.emit('info', message)
 				}, // onLog
-				async (payload: Buffer): Promise<void> => {
-					this._parseCommands(Buffer.from(payload))
+				async (payload: ThreadedPayload): Promise<void> => {
+					const normalizedPayload = this._normalizePayload(payload)
+					if (!normalizedPayload) {
+						this.emit('error', `Received invalid command payload type: ${typeof payload}`)
+						return
+					}
+
+					this._parseCommands(normalizedPayload)
 				}, // onCommandsReceived
 				async (ids: Array<{ packetId: number; trackingId: number }>): Promise<void> => {
 					this.emit(
@@ -179,12 +190,14 @@ export class AtemSocket extends EventEmitter<AtemSocketEvents> {
 
 	private _parseCommands(buffer: Buffer): IDeserializedCommand[] {
 		const parsedCommands: IDeserializedCommand[] = []
+		let isFirstCommand = true
+		let keepCache = false
 
 		while (buffer.length > 8) {
 			const length = buffer.readUInt16BE(0)
 			const name = buffer.toString('ascii', 4, 8)
 
-			if (length < 8) {
+			if (length < 8 || length > buffer.length) {
 				// Commands are never less than 8, as that is the header
 				break
 			}
@@ -202,6 +215,23 @@ export class AtemSocket extends EventEmitter<AtemSocketEvents> {
 						this._commandParser.version = cmd.properties.version
 					}
 
+					if (isFirstCommand) {
+						// If the first command is a TimeCommand, we need to check if the remaining data matches
+						// the last batch's remainder.
+						if (cmd instanceof TimeCommand) {
+							// Mark cache as valid for this batch, as the first command is a TimeCommand.
+							keepCache = true
+							const remainder = buffer.slice(length)
+							// Check if the remaining commands match the last batch, if so, skip them.
+							if (this._lastTimeCommandRemainder && remainder.equals(this._lastTimeCommandRemainder)) {
+								parsedCommands.push(cmd)
+								break
+							}
+							// Otherwise, cache the remainder for comparison with the next batch.
+							this._lastTimeCommandRemainder = Buffer.from(remainder)
+						}
+					}
+
 					parsedCommands.push(cmd)
 				} catch (e) {
 					this.emit('error', `Failed to deserialize command: ${cmdConstructor.constructor.name}: ${e}`)
@@ -210,13 +240,38 @@ export class AtemSocket extends EventEmitter<AtemSocketEvents> {
 				this.emit('debug', `Unknown command ${name} (${length}b)`)
 			}
 
+			// Always clear the first command flag after processing the first command.
+			isFirstCommand = false
+
 			// Trim the buffer
 			buffer = buffer.slice(length)
 		}
 
+		if (!keepCache) {
+			// Always clear the cache if the first command was not a TimeCommand,
+			// as the remainder could be invalid.
+			this._lastTimeCommandRemainder = undefined
+		}
+
 		if (parsedCommands.length > 0) {
 			this.emit('receivedCommands', parsedCommands)
 		}
 		return parsedCommands
 	}
+
+	private _normalizePayload(payload: ThreadedPayload): Buffer | undefined {
+		if (Buffer.isBuffer(payload)) {
+			return payload
+		}
+
+		if (payload instanceof Uint8Array) {
+			return Buffer.from(payload.buffer, payload.byteOffset, payload.byteLength)
+		}
+
+		if (payload && payload.type === 'Buffer' && Array.isArray(payload.data)) {
+			return Buffer.from(payload.data)
+		}
+
+		return undefined
+	}
 }