David Snelling
47ba4f4093
Made code more concise by removing unnecessary semicolons and adjusting the code formatting to align with the project's style. Improved readability and maintainability.
860 lines
25 KiB
TypeScript
860 lines
25 KiB
TypeScript
/**
|
|
* Augmentation Event Pipeline
|
|
*
|
|
* This module provides a pipeline for managing and executing multiple augmentations
|
|
* of each type. It allows registering multiple augmentations and executing them
|
|
* in sequence or in parallel.
|
|
*/
|
|
|
|
import {
|
|
BrainyAugmentations,
|
|
IAugmentation,
|
|
IWebSocketSupport,
|
|
AugmentationResponse,
|
|
AugmentationType
|
|
} from './types/augmentations.js'
|
|
import { isThreadingAvailable, isBrowser, isNode } from './utils/environment.js'
|
|
import { executeInThread } from './utils/workerUtils.js'
|
|
|
|
/**
|
|
* Type definitions for the augmentation registry
|
|
*/
|
|
type AugmentationRegistry = {
|
|
sense: BrainyAugmentations.ISenseAugmentation[]
|
|
conduit: BrainyAugmentations.IConduitAugmentation[]
|
|
cognition: BrainyAugmentations.ICognitionAugmentation[]
|
|
memory: BrainyAugmentations.IMemoryAugmentation[]
|
|
perception: BrainyAugmentations.IPerceptionAugmentation[]
|
|
dialog: BrainyAugmentations.IDialogAugmentation[]
|
|
activation: BrainyAugmentations.IActivationAugmentation[]
|
|
webSocket: IWebSocketSupport[]
|
|
}
|
|
|
|
/**
|
|
* Execution mode for the pipeline
|
|
*/
|
|
export enum ExecutionMode {
|
|
SEQUENTIAL = 'sequential',
|
|
PARALLEL = 'parallel',
|
|
FIRST_SUCCESS = 'firstSuccess',
|
|
FIRST_RESULT = 'firstResult',
|
|
THREADED = 'threaded' // Execute in separate threads when available
|
|
}
|
|
|
|
/**
|
|
* Options for pipeline execution
|
|
*/
|
|
export interface PipelineOptions {
|
|
mode?: ExecutionMode
|
|
timeout?: number
|
|
stopOnError?: boolean
|
|
forceThreading?: boolean // Force threading even if not in THREADED mode
|
|
disableThreading?: boolean // Disable threading even if in THREADED mode
|
|
}
|
|
|
|
/**
|
|
* Default pipeline options
|
|
*/
|
|
const DEFAULT_PIPELINE_OPTIONS: PipelineOptions = {
|
|
mode: ExecutionMode.SEQUENTIAL,
|
|
timeout: 30000,
|
|
stopOnError: false,
|
|
forceThreading: false,
|
|
disableThreading: false
|
|
}
|
|
|
|
/**
|
|
* AugmentationPipeline class
|
|
*
|
|
* Manages multiple augmentations of each type and provides methods to execute them.
|
|
*/
|
|
export class AugmentationPipeline {
|
|
private registry: AugmentationRegistry = {
|
|
sense: [],
|
|
conduit: [],
|
|
cognition: [],
|
|
memory: [],
|
|
perception: [],
|
|
dialog: [],
|
|
activation: [],
|
|
webSocket: []
|
|
}
|
|
|
|
/**
|
|
* Register an augmentation with the pipeline
|
|
*
|
|
* @param augmentation The augmentation to register
|
|
* @returns The pipeline instance for chaining
|
|
*/
|
|
public register<T extends IAugmentation>(
|
|
augmentation: T
|
|
): AugmentationPipeline {
|
|
let registered = false
|
|
|
|
// Check for specific augmentation types
|
|
if (
|
|
this.isAugmentationType<BrainyAugmentations.ISenseAugmentation>(
|
|
augmentation,
|
|
'processRawData',
|
|
'listenToFeed'
|
|
)
|
|
) {
|
|
this.registry.sense.push(augmentation)
|
|
registered = true
|
|
} else if (
|
|
this.isAugmentationType<BrainyAugmentations.IConduitAugmentation>(
|
|
augmentation,
|
|
'establishConnection',
|
|
'readData',
|
|
'writeData',
|
|
'monitorStream'
|
|
)
|
|
) {
|
|
this.registry.conduit.push(augmentation)
|
|
registered = true
|
|
} else if (
|
|
this.isAugmentationType<BrainyAugmentations.ICognitionAugmentation>(
|
|
augmentation,
|
|
'reason',
|
|
'infer',
|
|
'executeLogic'
|
|
)
|
|
) {
|
|
this.registry.cognition.push(augmentation)
|
|
registered = true
|
|
} else if (
|
|
this.isAugmentationType<BrainyAugmentations.IMemoryAugmentation>(
|
|
augmentation,
|
|
'storeData',
|
|
'retrieveData',
|
|
'updateData',
|
|
'deleteData',
|
|
'listDataKeys'
|
|
)
|
|
) {
|
|
this.registry.memory.push(augmentation)
|
|
registered = true
|
|
} else if (
|
|
this.isAugmentationType<BrainyAugmentations.IPerceptionAugmentation>(
|
|
augmentation,
|
|
'interpret',
|
|
'organize',
|
|
'generateVisualization'
|
|
)
|
|
) {
|
|
this.registry.perception.push(augmentation)
|
|
registered = true
|
|
} else if (
|
|
this.isAugmentationType<BrainyAugmentations.IDialogAugmentation>(
|
|
augmentation,
|
|
'processUserInput',
|
|
'generateResponse',
|
|
'manageContext'
|
|
)
|
|
) {
|
|
this.registry.dialog.push(augmentation)
|
|
registered = true
|
|
} else if (
|
|
this.isAugmentationType<BrainyAugmentations.IActivationAugmentation>(
|
|
augmentation,
|
|
'triggerAction',
|
|
'generateOutput',
|
|
'interactExternal'
|
|
)
|
|
) {
|
|
this.registry.activation.push(augmentation)
|
|
registered = true
|
|
}
|
|
|
|
// Check if the augmentation supports WebSocket
|
|
if (
|
|
this.isAugmentationType<IWebSocketSupport>(
|
|
augmentation,
|
|
'connectWebSocket',
|
|
'sendWebSocketMessage',
|
|
'onWebSocketMessage',
|
|
'closeWebSocket'
|
|
)
|
|
) {
|
|
this.registry.webSocket.push(augmentation as IWebSocketSupport)
|
|
registered = true
|
|
}
|
|
|
|
// If the augmentation wasn't registered as any known type, throw an error
|
|
if (!registered) {
|
|
throw new Error(`Unknown augmentation type: ${augmentation.name}`)
|
|
}
|
|
|
|
return this
|
|
}
|
|
|
|
/**
|
|
* Unregister an augmentation from the pipeline
|
|
*
|
|
* @param augmentationName The name of the augmentation to unregister
|
|
* @returns The pipeline instance for chaining
|
|
*/
|
|
public unregister(augmentationName: string): AugmentationPipeline {
|
|
let found = false
|
|
|
|
// Remove from all registries
|
|
for (const type in this.registry) {
|
|
const typedRegistry = this.registry[type as keyof AugmentationRegistry]
|
|
const index = typedRegistry.findIndex(
|
|
(aug) => aug.name === augmentationName
|
|
)
|
|
|
|
if (index !== -1) {
|
|
typedRegistry.splice(index, 1)
|
|
found = true
|
|
}
|
|
}
|
|
|
|
return this
|
|
}
|
|
|
|
/**
|
|
* Initialize all registered augmentations
|
|
*
|
|
* @returns A promise that resolves when all augmentations are initialized
|
|
*/
|
|
public async initialize(): Promise<void> {
|
|
const allAugmentations = this.getAllAugmentations()
|
|
|
|
await Promise.all(
|
|
allAugmentations.map((augmentation) =>
|
|
augmentation.initialize().catch((error) => {
|
|
console.error(
|
|
`Failed to initialize augmentation ${augmentation.name}:`,
|
|
error
|
|
)
|
|
})
|
|
)
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Shut down all registered augmentations
|
|
*
|
|
* @returns A promise that resolves when all augmentations are shut down
|
|
*/
|
|
public async shutDown(): Promise<void> {
|
|
const allAugmentations = this.getAllAugmentations()
|
|
|
|
await Promise.all(
|
|
allAugmentations.map((augmentation) =>
|
|
augmentation.shutDown().catch((error) => {
|
|
console.error(
|
|
`Failed to shut down augmentation ${augmentation.name}:`,
|
|
error
|
|
)
|
|
})
|
|
)
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Execute a sense pipeline
|
|
*
|
|
* @param method The method to execute on each sense augmentation
|
|
* @param args The arguments to pass to the method
|
|
* @param options The pipeline execution options
|
|
* @returns A promise that resolves with the results from all augmentations
|
|
*/
|
|
public async executeSensePipeline<
|
|
M extends keyof BrainyAugmentations.ISenseAugmentation & string,
|
|
R extends BrainyAugmentations.ISenseAugmentation[M] extends (
|
|
...args: any[]
|
|
) => AugmentationResponse<infer U>
|
|
? U
|
|
: never
|
|
>(
|
|
method: M &
|
|
(BrainyAugmentations.ISenseAugmentation[M] extends (...args: any[]) => any
|
|
? M
|
|
: never),
|
|
args: Parameters<
|
|
Extract<
|
|
BrainyAugmentations.ISenseAugmentation[M],
|
|
(...args: any[]) => any
|
|
>
|
|
>,
|
|
options: PipelineOptions = {}
|
|
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
|
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
|
|
return this.executeTypedPipeline<
|
|
BrainyAugmentations.ISenseAugmentation,
|
|
M,
|
|
R
|
|
>(this.registry.sense, method, args, opts)
|
|
}
|
|
|
|
/**
|
|
* Execute a conduit pipeline
|
|
*
|
|
* @param method The method to execute on each conduit augmentation
|
|
* @param args The arguments to pass to the method
|
|
* @param options The pipeline execution options
|
|
* @returns A promise that resolves with the results from all augmentations
|
|
*/
|
|
public async executeConduitPipeline<
|
|
M extends keyof BrainyAugmentations.IConduitAugmentation & string,
|
|
R extends BrainyAugmentations.IConduitAugmentation[M] extends (
|
|
...args: any[]
|
|
) => AugmentationResponse<infer U>
|
|
? U
|
|
: never
|
|
>(
|
|
method: M &
|
|
(BrainyAugmentations.IConduitAugmentation[M] extends (
|
|
...args: any[]
|
|
) => any
|
|
? M
|
|
: never),
|
|
args: Parameters<
|
|
Extract<
|
|
BrainyAugmentations.IConduitAugmentation[M],
|
|
(...args: any[]) => any
|
|
>
|
|
>,
|
|
options: PipelineOptions = {}
|
|
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
|
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
|
|
return this.executeTypedPipeline<
|
|
BrainyAugmentations.IConduitAugmentation,
|
|
M,
|
|
R
|
|
>(this.registry.conduit, method, args, opts)
|
|
}
|
|
|
|
/**
|
|
* Execute a cognition pipeline
|
|
*
|
|
* @param method The method to execute on each cognition augmentation
|
|
* @param args The arguments to pass to the method
|
|
* @param options The pipeline execution options
|
|
* @returns A promise that resolves with the results from all augmentations
|
|
*/
|
|
public async executeCognitionPipeline<
|
|
M extends keyof BrainyAugmentations.ICognitionAugmentation & string,
|
|
R extends BrainyAugmentations.ICognitionAugmentation[M] extends (
|
|
...args: any[]
|
|
) => AugmentationResponse<infer U>
|
|
? U
|
|
: never
|
|
>(
|
|
method: M &
|
|
(BrainyAugmentations.ICognitionAugmentation[M] extends (
|
|
...args: any[]
|
|
) => any
|
|
? M
|
|
: never),
|
|
args: Parameters<
|
|
Extract<
|
|
BrainyAugmentations.ICognitionAugmentation[M],
|
|
(...args: any[]) => any
|
|
>
|
|
>,
|
|
options: PipelineOptions = {}
|
|
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
|
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
|
|
return this.executeTypedPipeline<
|
|
BrainyAugmentations.ICognitionAugmentation,
|
|
M,
|
|
R
|
|
>(this.registry.cognition, method, args, opts)
|
|
}
|
|
|
|
/**
|
|
* Execute a memory pipeline
|
|
*
|
|
* @param method The method to execute on each memory augmentation
|
|
* @param args The arguments to pass to the method
|
|
* @param options The pipeline execution options
|
|
* @returns A promise that resolves with the results from all augmentations
|
|
*/
|
|
public async executeMemoryPipeline<
|
|
M extends keyof BrainyAugmentations.IMemoryAugmentation & string,
|
|
R extends BrainyAugmentations.IMemoryAugmentation[M] extends (
|
|
...args: any[]
|
|
) => AugmentationResponse<infer U>
|
|
? U
|
|
: never
|
|
>(
|
|
method: M &
|
|
(BrainyAugmentations.IMemoryAugmentation[M] extends (
|
|
...args: any[]
|
|
) => any
|
|
? M
|
|
: never),
|
|
args: Parameters<
|
|
Extract<
|
|
BrainyAugmentations.IMemoryAugmentation[M],
|
|
(...args: any[]) => any
|
|
>
|
|
>,
|
|
options: PipelineOptions = {}
|
|
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
|
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
|
|
return this.executeTypedPipeline<
|
|
BrainyAugmentations.IMemoryAugmentation,
|
|
M,
|
|
R
|
|
>(this.registry.memory, method, args, opts)
|
|
}
|
|
|
|
/**
|
|
* Execute a perception pipeline
|
|
*
|
|
* @param method The method to execute on each perception augmentation
|
|
* @param args The arguments to pass to the method
|
|
* @param options The pipeline execution options
|
|
* @returns A promise that resolves with the results from all augmentations
|
|
*/
|
|
public async executePerceptionPipeline<
|
|
M extends keyof BrainyAugmentations.IPerceptionAugmentation & string,
|
|
R extends BrainyAugmentations.IPerceptionAugmentation[M] extends (
|
|
...args: any[]
|
|
) => AugmentationResponse<infer U>
|
|
? U
|
|
: never
|
|
>(
|
|
method: M &
|
|
(BrainyAugmentations.IPerceptionAugmentation[M] extends (
|
|
...args: any[]
|
|
) => any
|
|
? M
|
|
: never),
|
|
args: Parameters<
|
|
Extract<
|
|
BrainyAugmentations.IPerceptionAugmentation[M],
|
|
(...args: any[]) => any
|
|
>
|
|
>,
|
|
options: PipelineOptions = {}
|
|
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
|
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
|
|
return this.executeTypedPipeline<
|
|
BrainyAugmentations.IPerceptionAugmentation,
|
|
M,
|
|
R
|
|
>(this.registry.perception, method, args, opts)
|
|
}
|
|
|
|
/**
|
|
* Execute a dialog pipeline
|
|
*
|
|
* @param method The method to execute on each dialog augmentation
|
|
* @param args The arguments to pass to the method
|
|
* @param options The pipeline execution options
|
|
* @returns A promise that resolves with the results from all augmentations
|
|
*/
|
|
public async executeDialogPipeline<
|
|
M extends keyof BrainyAugmentations.IDialogAugmentation & string,
|
|
R extends BrainyAugmentations.IDialogAugmentation[M] extends (
|
|
...args: any[]
|
|
) => AugmentationResponse<infer U>
|
|
? U
|
|
: never
|
|
>(
|
|
method: M &
|
|
(BrainyAugmentations.IDialogAugmentation[M] extends (
|
|
...args: any[]
|
|
) => any
|
|
? M
|
|
: never),
|
|
args: Parameters<
|
|
Extract<
|
|
BrainyAugmentations.IDialogAugmentation[M],
|
|
(...args: any[]) => any
|
|
>
|
|
>,
|
|
options: PipelineOptions = {}
|
|
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
|
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
|
|
return this.executeTypedPipeline<
|
|
BrainyAugmentations.IDialogAugmentation,
|
|
M,
|
|
R
|
|
>(this.registry.dialog, method, args, opts)
|
|
}
|
|
|
|
/**
|
|
* Execute an activation pipeline
|
|
*
|
|
* @param method The method to execute on each activation augmentation
|
|
* @param args The arguments to pass to the method
|
|
* @param options The pipeline execution options
|
|
* @returns A promise that resolves with the results from all augmentations
|
|
*/
|
|
public async executeActivationPipeline<
|
|
M extends keyof BrainyAugmentations.IActivationAugmentation & string,
|
|
R extends BrainyAugmentations.IActivationAugmentation[M] extends (
|
|
...args: any[]
|
|
) => AugmentationResponse<infer U>
|
|
? U
|
|
: never
|
|
>(
|
|
method: M &
|
|
(BrainyAugmentations.IActivationAugmentation[M] extends (
|
|
...args: any[]
|
|
) => any
|
|
? M
|
|
: never),
|
|
args: Parameters<
|
|
Extract<
|
|
BrainyAugmentations.IActivationAugmentation[M],
|
|
(...args: any[]) => any
|
|
>
|
|
>,
|
|
options: PipelineOptions = {}
|
|
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
|
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
|
|
return this.executeTypedPipeline<
|
|
BrainyAugmentations.IActivationAugmentation,
|
|
M,
|
|
R
|
|
>(this.registry.activation, method, args, opts)
|
|
}
|
|
|
|
/**
|
|
* Get all registered augmentations
|
|
*
|
|
* @returns An array of all registered augmentations
|
|
*/
|
|
public getAllAugmentations(): IAugmentation[] {
|
|
// Create a Set to avoid duplicates (an augmentation might be in multiple registries)
|
|
const allAugmentations = new Set<IAugmentation>([
|
|
...this.registry.sense,
|
|
...this.registry.conduit,
|
|
...this.registry.cognition,
|
|
...this.registry.memory,
|
|
...this.registry.perception,
|
|
...this.registry.dialog,
|
|
...this.registry.activation,
|
|
...this.registry.webSocket
|
|
])
|
|
|
|
// Convert back to array
|
|
return Array.from(allAugmentations)
|
|
}
|
|
|
|
/**
|
|
* Get all augmentations of a specific type
|
|
*
|
|
* @param type The type of augmentation to get
|
|
* @returns An array of all augmentations of the specified type
|
|
*/
|
|
public getAugmentationsByType(type: AugmentationType): IAugmentation[] {
|
|
switch (type) {
|
|
case AugmentationType.SENSE:
|
|
return [...this.registry.sense]
|
|
case AugmentationType.CONDUIT:
|
|
return [...this.registry.conduit]
|
|
case AugmentationType.COGNITION:
|
|
return [...this.registry.cognition]
|
|
case AugmentationType.MEMORY:
|
|
return [...this.registry.memory]
|
|
case AugmentationType.PERCEPTION:
|
|
return [...this.registry.perception]
|
|
case AugmentationType.DIALOG:
|
|
return [...this.registry.dialog]
|
|
case AugmentationType.ACTIVATION:
|
|
return [...this.registry.activation]
|
|
case AugmentationType.WEBSOCKET:
|
|
return [...this.registry.webSocket]
|
|
default:
|
|
return []
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get all available augmentation types
|
|
*
|
|
* @returns An array of all augmentation types that have at least one registered augmentation
|
|
*/
|
|
public getAvailableAugmentationTypes(): AugmentationType[] {
|
|
const availableTypes: AugmentationType[] = []
|
|
|
|
if (this.registry.sense.length > 0)
|
|
availableTypes.push(AugmentationType.SENSE)
|
|
if (this.registry.conduit.length > 0)
|
|
availableTypes.push(AugmentationType.CONDUIT)
|
|
if (this.registry.cognition.length > 0)
|
|
availableTypes.push(AugmentationType.COGNITION)
|
|
if (this.registry.memory.length > 0)
|
|
availableTypes.push(AugmentationType.MEMORY)
|
|
if (this.registry.perception.length > 0)
|
|
availableTypes.push(AugmentationType.PERCEPTION)
|
|
if (this.registry.dialog.length > 0)
|
|
availableTypes.push(AugmentationType.DIALOG)
|
|
if (this.registry.activation.length > 0)
|
|
availableTypes.push(AugmentationType.ACTIVATION)
|
|
if (this.registry.webSocket.length > 0)
|
|
availableTypes.push(AugmentationType.WEBSOCKET)
|
|
|
|
return availableTypes
|
|
}
|
|
|
|
/**
|
|
* Get all WebSocket-supporting augmentations
|
|
*
|
|
* @returns An array of all augmentations that support WebSocket connections
|
|
*/
|
|
public getWebSocketAugmentations(): IWebSocketSupport[] {
|
|
return [...this.registry.webSocket]
|
|
}
|
|
|
|
/**
|
|
* Check if an augmentation is of a specific type
|
|
*
|
|
* @param augmentation The augmentation to check
|
|
* @param methods The methods that should be present on the augmentation
|
|
* @returns True if the augmentation is of the specified type
|
|
*/
|
|
private isAugmentationType<T extends IAugmentation>(
|
|
augmentation: IAugmentation,
|
|
...methods: (keyof T)[]
|
|
): augmentation is T {
|
|
// First check that the augmentation has all the required base methods
|
|
const baseMethodsExist = ['initialize', 'shutDown', 'getStatus'].every(
|
|
(method) => typeof (augmentation as any)[method] === 'function'
|
|
)
|
|
|
|
if (!baseMethodsExist) {
|
|
return false
|
|
}
|
|
|
|
// Then check that it has all the specific methods for this type
|
|
return methods.every(
|
|
(method) => typeof (augmentation as any)[method] === 'function'
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Determines if threading should be used based on options and environment
|
|
*
|
|
* @param options The pipeline options
|
|
* @returns True if threading should be used, false otherwise
|
|
*/
|
|
private shouldUseThreading(options: PipelineOptions): boolean {
|
|
// If threading is explicitly disabled, don't use it
|
|
if (options.disableThreading) {
|
|
return false
|
|
}
|
|
|
|
// If threading is explicitly forced, use it if available
|
|
if (options.forceThreading) {
|
|
return isThreadingAvailable()
|
|
}
|
|
|
|
// If in THREADED mode, use threading if available
|
|
if (options.mode === ExecutionMode.THREADED) {
|
|
return isThreadingAvailable()
|
|
}
|
|
|
|
// Otherwise, don't use threading
|
|
return false
|
|
}
|
|
|
|
/**
|
|
* Execute a pipeline for a specific augmentation type
|
|
*
|
|
* @param augmentations The augmentations to execute
|
|
* @param method The method to execute on each augmentation
|
|
* @param args The arguments to pass to the method
|
|
* @param options The pipeline execution options
|
|
* @returns A promise that resolves with the results from all augmentations
|
|
*/
|
|
private async executeTypedPipeline<
|
|
T extends IAugmentation,
|
|
M extends keyof T & string,
|
|
R extends T[M] extends (...args: any[]) => AugmentationResponse<infer U>
|
|
? U
|
|
: never
|
|
>(
|
|
augmentations: T[],
|
|
method: M & (T[M] extends (...args: any[]) => any ? M : never),
|
|
args: Parameters<Extract<T[M], (...args: any[]) => any>>,
|
|
options: PipelineOptions
|
|
): Promise<
|
|
Promise<{
|
|
success: boolean
|
|
data: R
|
|
error?: string
|
|
}>[]
|
|
> {
|
|
// Filter out disabled augmentations
|
|
const enabledAugmentations = augmentations.filter(
|
|
(aug) => aug.enabled !== false
|
|
)
|
|
|
|
if (enabledAugmentations.length === 0) {
|
|
return []
|
|
}
|
|
|
|
// Create a function to execute the method on an augmentation
|
|
const executeMethod = async (
|
|
augmentation: T
|
|
): Promise<{
|
|
success: boolean
|
|
data: R
|
|
error?: string
|
|
}> => {
|
|
try {
|
|
// Create a timeout promise if a timeout is specified
|
|
const timeoutPromise = options.timeout
|
|
? new Promise<{
|
|
success: boolean
|
|
data: R
|
|
error?: string
|
|
}>((_, reject) => {
|
|
setTimeout(() => {
|
|
reject(
|
|
new Error(
|
|
`Timeout executing ${String(method)} on ${augmentation.name}`
|
|
)
|
|
)
|
|
}, options.timeout)
|
|
})
|
|
: null
|
|
|
|
// Check if threading should be used
|
|
const useThreading = this.shouldUseThreading(options)
|
|
|
|
// Execute the method on the augmentation, using threading if appropriate
|
|
let methodPromise: Promise<AugmentationResponse<R>>
|
|
|
|
if (useThreading) {
|
|
// Execute in a separate thread
|
|
try {
|
|
// Create a function that can be serialized and executed in a worker
|
|
const workerFn = (...workerArgs: any[]) => {
|
|
// This function will be stringified and executed in the worker
|
|
// It needs to be self-contained
|
|
const augFn = augmentation[method as string] as Function
|
|
return augFn.apply(augmentation, workerArgs)
|
|
}
|
|
|
|
methodPromise = executeInThread<AugmentationResponse<R>>(
|
|
workerFn.toString(),
|
|
args
|
|
)
|
|
} catch (threadError) {
|
|
console.warn(
|
|
`Failed to execute in thread, falling back to main thread: ${threadError}`
|
|
)
|
|
// Fall back to executing in the main thread
|
|
methodPromise = Promise.resolve(
|
|
(augmentation[method] as Function)(
|
|
...args
|
|
) as AugmentationResponse<R>
|
|
)
|
|
}
|
|
} else {
|
|
// Execute in the main thread
|
|
methodPromise = Promise.resolve(
|
|
(augmentation[method] as Function)(
|
|
...args
|
|
) as AugmentationResponse<R>
|
|
)
|
|
}
|
|
|
|
// Race the method promise against the timeout promise if a timeout is specified
|
|
const result = timeoutPromise
|
|
? await Promise.race([methodPromise, timeoutPromise])
|
|
: await methodPromise
|
|
|
|
return result
|
|
} catch (error) {
|
|
console.error(
|
|
`Error executing ${String(method)} on ${augmentation.name}:`,
|
|
error
|
|
)
|
|
return {
|
|
success: false,
|
|
data: null as unknown as R,
|
|
error: error instanceof Error ? error.message : String(error)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Execute the pipeline based on the specified mode
|
|
switch (options.mode) {
|
|
case ExecutionMode.PARALLEL:
|
|
// Execute all augmentations in parallel
|
|
return enabledAugmentations.map(executeMethod)
|
|
|
|
case ExecutionMode.THREADED:
|
|
// Execute all augmentations in parallel with threading enabled
|
|
// Force threading for this mode
|
|
const threadedOptions = { ...options, forceThreading: true }
|
|
|
|
// Create a new executeMethod function that uses the threaded options
|
|
const executeMethodThreaded = async (augmentation: T) => {
|
|
// Save the original options
|
|
const originalOptions = options
|
|
|
|
// Set the options to the threaded options
|
|
options = threadedOptions
|
|
|
|
// Execute the method
|
|
const result = await executeMethod(augmentation)
|
|
|
|
// Restore the original options
|
|
options = originalOptions
|
|
|
|
return result
|
|
}
|
|
|
|
return enabledAugmentations.map(executeMethodThreaded)
|
|
|
|
case ExecutionMode.FIRST_SUCCESS:
|
|
// Execute augmentations sequentially until one succeeds
|
|
for (const augmentation of enabledAugmentations) {
|
|
const resultPromise = executeMethod(augmentation)
|
|
const result = await resultPromise
|
|
if (result.success) {
|
|
return [resultPromise]
|
|
}
|
|
}
|
|
return []
|
|
|
|
case ExecutionMode.FIRST_RESULT:
|
|
// Execute augmentations sequentially until one returns a result
|
|
for (const augmentation of enabledAugmentations) {
|
|
const resultPromise = executeMethod(augmentation)
|
|
const result = await resultPromise
|
|
if (result.success && result.data) {
|
|
return [resultPromise]
|
|
}
|
|
}
|
|
return []
|
|
|
|
case ExecutionMode.SEQUENTIAL:
|
|
default:
|
|
// Execute augmentations sequentially
|
|
const results: Promise<{
|
|
success: boolean
|
|
data: R
|
|
error?: string
|
|
}>[] = []
|
|
for (const augmentation of enabledAugmentations) {
|
|
const resultPromise = executeMethod(augmentation)
|
|
results.push(resultPromise)
|
|
|
|
// Check if we need to stop on error
|
|
if (options.stopOnError) {
|
|
const result = await resultPromise
|
|
if (!result.success) {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
return results
|
|
}
|
|
}
|
|
}
|
|
|
|
// Create and export a default instance of the pipeline
|
|
export const augmentationPipeline = new AugmentationPipeline()
|