Fix dynamic, self-adjusting priority queue

I’m trying to implement a dynamic, self-adjusting priority queue in JavaScript that operates with asynchronous tasks. The goal is to create a priority queue where tasks can be added at any time with a given priority. However, the complexity arises from the requirement that the priority of tasks can change dynamically based on external asynchronous events.

Here are the specific requirements:

Dynamic Task Addition: The queue should allow adding tasks (function objects) with a specified priority at any time.

Asynchronous Task Execution: Tasks should be executed asynchronously using async/await.

Priority Adjustment: There should be a way to adjust the priority of tasks already in the queue, based on external asynchronous triggers (like fetching data from an API). The queue should automatically reorder tasks based on new priorities.

Concurrency Control: The system should process up to N tasks concurrently, where N is a configurable parameter.

Event Handling: The system should emit an event after the completion of each task and handle errors gracefully, without stopping the queue processing.

Task Cancellation: Provide a mechanism to cancel a queued task.

I’ve done some preliminary research and understand that this will likely involve a combination of async/await, Promises, and perhaps some form of event handling or reactive programming, but I’m struggling to figure out how to architect this system in a scalable and efficient way.

Could someone provide guidance or a reference implementation for such a system? Any help or pointers towards relevant patterns or libraries would be greatly appreciated.

class AsyncPriorityQueue {
    constructor() {
        this.queue = [];
        this.processing = false;
    }

    // Add task to the queue with a given priority
    add(task, priority) {
        this.queue.push({ task, priority });
        this.queue.sort((a, b) => a.priority - b.priority);
        this.processQueue();
    }

    // Process the queue
    async processQueue() {
        if (this.processing || this.queue.length === 0) {
            return;
        }
        this.processing = true;

        const { task } = this.queue.shift();
        try {
            await task();
        } catch (error) {
            console.error("Task failed:", error);
        }

        this.processing = false;
        this.processQueue();
    }

    
    adjustPriority(taskId, newPriority) {}
}

// Usage Example
const myQueue = new AsyncPriorityQueue();

// Example tasks
const task1 = () => new Promise(resolve => setTimeout(resolve, 1000, 'Task 1'));
const task2 = () => new Promise(resolve => setTimeout(resolve, 1000, 'Task 2'));

myQueue.add(task1, 2);
myQueue.add(task2, 1);

>Solution :

This implementation addresses the concurrency control by allowing multiple tasks to be processed in parallel (up to the limit set by concurrency), dynamically adjusts task priorities, and provides a way to cancel tasks. The queue is always sorted before a new task is dequeued, ensuring that tasks are processed in the correct priority order.

class FixedAsyncPriorityQueue {
    constructor(concurrency = 1) {
        this.queue = [];
        this.concurrency = concurrency;
        this.currentTasks = new Set();
        this.taskCounter = 0;
    }

    // Add task to the queue with a given priority
    add(task, priority) {
        const taskId = this.taskCounter++;
        this.queue.push({ task, taskId, priority });
        this.processQueue();
        return taskId;
    }

    // Re-sort the queue when priorities change
    sortQueue() {
        this.queue.sort((a, b) => a.priority - b.priority);
    }

    // Process the queue with concurrency control
    async processQueue() {
        while (this.queue.length > 0 && this.currentTasks.size < this.concurrency) {
            this.sortQueue();
            const { task, taskId } = this.queue.shift();
            this.currentTasks.add(taskId);

            task().then(() => {
                this.currentTasks.delete(taskId);
                this.processQueue();
            }).catch(error => {
                console.error(`Task ${taskId} failed:`, error);
                this.currentTasks.delete(taskId);
                this.processQueue();
            });
        }
    }

    // Adjust the priority of a specific task
    adjustPriority(taskId, newPriority) {
        const taskIndex = this.queue.findIndex(task => task.taskId === taskId);
        if (taskIndex > -1) {
            this.queue[taskIndex].priority = newPriority;
            this.sortQueue();
        }
    }

    // Cancel a task
    cancelTask(taskId) {
        const taskIndex = this.queue.findIndex(task => task.taskId === taskId);
        if (taskIndex > -1) {
            this.queue.splice(taskIndex, 1);
        }
    }
}

// Usage example
const myQueue = new FixedAsyncPriorityQueue(2); // Set concurrency to 2

// Example tasks
const task1 = () => new Promise(resolve => setTimeout(resolve, 1000, 'Task 1'));
const task2 = () => new Promise(resolve => setTimeout(resolve, 1000, 'Task 2'));

const taskId1 = myQueue.add(task1, 2);
const taskId2 = myQueue.add(task2, 1);