import { Node } from "@counterfactual/types" ; import Queue , { Task } from "p-queue" ; import { addToManyQueues } from "./methods/queued-execution" ; class QueueWithLockingServiceConnection extends Queue { constructor ( private readonly lockName , private readonly lockingService , ... args ) { super ( ... args ) ; } async add ( task ) { return super . add ( ( ) => this . lockingService . acquireLock ( this . lockName , task , 30000 ) ) ; } } export default class ProcessQueue { private readonly queues : Map < string , QueueWithLockingServiceConnection | Queue > = new Map < string , QueueWithLockingServiceConnection | Queue > ( ) ; constructor ( private readonly lockingService ? ) { } addTask ( lockNames , task ) { return addToManyQueues ( lockNames . map ( this . getOrCreateLockQueue . bind ( this ) ) , task ) ; } private getOrCreateLockQueue ( lockName ) { if ( ! this . queues . has ( lockName ) ) { this . queues . set ( lockName , this . lockingService ? new QueueWithLockingServiceConnection ( lockName , this . lockingService , { concurrency : 1 } ) : new Queue ( { concurrency : 1 } ) ) ; } return this . queues . get ( lockName ) ! ; } }