Merge pull request #346 from nrkno/fix/http-queueId/sofie-3431

fix(HTTPSend): Allow HTTPSend commands to be sent in separate sequential queues (SOFIE-3431)
nrkno · Sep 9, 2024 · 44e9f5a · 44e9f5a
2 parents 81af9f7 + 5797d0d
commit 44e9f5a
Show file tree

Hide file tree

Showing 6 changed files with 29 additions and 15 deletions.
diff --git a/packages/timeline-state-resolver-types/src/index.ts b/packages/timeline-state-resolver-types/src/index.ts
@@ -170,5 +170,6 @@ export interface ActionExecutionResult<ResultData = undefined> {
 
 export enum ActionExecutionResultCode {
 	Error = 'ERROR',
+	IgnoredNotRelevant = 'IGNORED',
 	Ok = 'OK',
 }
diff --git a/packages/timeline-state-resolver/src/integrations/httpSend/__tests__/httpsend.spec.ts b/packages/timeline-state-resolver/src/integrations/httpSend/__tests__/httpsend.spec.ts
@@ -346,6 +346,7 @@ describe('HTTP-Send', () => {
 							content: content,
 							layer: 'layer0',
 						},
+						queueId: undefined,
 					},
 				])
 				await Promise.all(commands.map(async (c) => device.sendCommand(c)))
@@ -362,6 +363,7 @@ describe('HTTP-Send', () => {
 							content: content,
 							layer: 'layer0',
 						},
+						queueId: undefined,
 					},
 				])
 			}

diff --git a/packages/timeline-state-resolver/src/integrations/httpSend/index.ts b/packages/timeline-state-resolver/src/integrations/httpSend/index.ts
@@ -136,6 +136,7 @@ export class HTTPSendDevice extends Device<HTTPSendOptions, HttpSendDeviceState,
 						content: newLayer.content as HTTPSendCommandContent,
 						layer: layerKey,
 					},
+					queueId: (newLayer.content as HTTPSendCommandContent)?.queueId,
 				})
 			} else {
 				// changed?
@@ -149,6 +150,7 @@ export class HTTPSendDevice extends Device<HTTPSendOptions, HttpSendDeviceState,
 							content: newLayer.content as HTTPSendCommandContent,
 							layer: layerKey,
 						},
+						queueId: (newLayer.content as HTTPSendCommandContent)?.queueId,
 					})
 				}
 			}
@@ -162,6 +164,7 @@ export class HTTPSendDevice extends Device<HTTPSendOptions, HttpSendDeviceState,
 					timelineObjId: oldLayer.id,
 					context: `removed: ${oldLayer.id}`,
 					command: { commandName: 'removed', content: oldLayer.content as HTTPSendCommandContent, layer: layerKey },
+					queueId: (oldLayer.content as HTTPSendCommandContent)?.queueId,
 				})
 			}
 		})
@@ -194,7 +197,7 @@ export class HTTPSendDevice extends Device<HTTPSendOptions, HttpSendDeviceState,
 			const trackedHash = this.trackedState.get(command.layer)
 			if (commandHash !== trackedHash)
 				return {
-					result: ActionExecutionResultCode.Error,
+					result: ActionExecutionResultCode.IgnoredNotRelevant,
 				} // command is no longer relevant to state
 		}
 		if (this._terminated) {
@@ -254,7 +257,7 @@ export class HTTPSendDevice extends Device<HTTPSendOptions, HttpSendDeviceState,
 
 			const response = await httpReq(command.content.url, options)
 
-			if (response.statusCode === 200) {
+			if (response.statusCode >= 200 && response.statusCode < 300) {
 				this.context.logger.debug(
 					`HTTPSend: ${command.content.type}: Good statuscode response on url "${command.content.url}": ${response.statusCode} (${context})`
 				)

diff --git a/packages/timeline-state-resolver/src/integrations/quantel/connection.ts b/packages/timeline-state-resolver/src/integrations/quantel/connection.ts
@@ -591,11 +591,10 @@ export class QuantelManager extends EventEmitter {
 		})
 	}
 	public clearAllWaitWithPort(portId: string) {
-		if (!this._waitWithPorts[portId]) {
-			_.each(this._waitWithPorts[portId], (fcn) => {
-				fcn(true)
-			})
-		}
+		if (!this._waitWithPorts[portId]) return
+		_.each(this._waitWithPorts[portId], (fcn) => {
+			fcn(true)
+		})
 	}
 	/**
 	 * Returns true if the wait was cleared from someone else

diff --git a/packages/timeline-state-resolver/src/service/commandExecutor.ts b/packages/timeline-state-resolver/src/service/commandExecutor.ts
@@ -1,3 +1,4 @@
+import * as _ from 'underscore'
 import { BaseDeviceAPI, CommandWithContext } from './device'
 import { Measurement } from './measure'
 import { StateHandlerContext } from './stateHandler'
@@ -50,15 +51,21 @@ export class CommandExecutor<DeviceState, Command extends CommandWithContext> {
 	): Promise<void> {
 		const start = Date.now() // note - would be better to use monotonic time here but BigInt's are annoying
 
-		for (const command of commands || []) {
-			const timeToWait = totalTime - (Date.now() - start)
-			if (timeToWait > 0) await wait(timeToWait)
+		const commandQueues = _.groupBy(commands || [], (command) => command.queueId ?? '$$default')
 
-			measurement?.executeCommand(command)
-			await this.sendCommand(command).catch((e) => {
-				this.logger.error('Error while executing command', e)
+		await Promise.allSettled(
+			Object.values<Command[]>(commandQueues).map(async (commandsInQueue): Promise<void> => {
+				for (const command of commandsInQueue) {
+					const timeToWait = totalTime - (Date.now() - start)
+					if (timeToWait > 0) await wait(timeToWait)
+
+					measurement?.executeCommand(command)
+					await this.sendCommand(command).catch((e) => {
+						this.logger.error('Error while executing command', e)
+					})
+					measurement?.finishedCommandExecution(command)
+				}
 			})
-			measurement?.finishedCommandExecution(command)
-		}
+		)
 	}
 }
diff --git a/packages/timeline-state-resolver/src/service/device.ts b/packages/timeline-state-resolver/src/service/device.ts
@@ -18,6 +18,8 @@ export type CommandWithContext = {
 	timelineObjId: string
 	/** this command is to be executed x ms _before_ the scheduled time */
 	preliminary?: number
+	/** commands with different queueId's can be executed in parallel in sequential mode */
+	queueId?: string
 }
 
 /**