@hao-rxtool/queue

This library provides a flexible and easy-to-use stream queue implementation for managing asynchronous tasks.

The library is built on top of RxJS, a powerful library for reactive programming.

Installation

Install the library using npm:

1npm install rxjs @hao-rxtool/queue

StreamQueue

Overview

The StreamQueue class represents a queue that allows enqueueing, dequeuing, and notifying subscribers of its status changes.

It provides a set of observables and methods for interacting with the queue.

Usage

1import { StreamQueue, StreamQueueStatus } from '@hao-rxtool/queue';
2
3// Create a new stream queue
4const queue = new StreamQueue<number>();
5
6// Listen for the status of the queue changes
7queue.statusChange.subscribe((status) => {
8  console.log(`Queue status: ${status}`); // StreamQueueStatus
9});
10
11// Listen for an item is ready to be dequeued
12queue.whenReadyToDequeue.subscribe((item) => {
13  console.log(`When ready to dequeue item: ${item}`);
14});
15
16// Listen for an item has been dequeued
17queue.whenDequeued.subscribe((item) => {
18  console.log(`Dequeued item: ${item}`);
19});
20
21// Enqueue an item
22queue.enqueue(1);
23
24// Dequeue an item
25queue.dequeue();
26
27// Enqueues an item and waits for it to be dequeued
28queue.enqueueAndWaitDequeue(2).subscribe((item) => {
29  console.log(`Enqueue and wait dequeue item: ${item}`);
30});
31
32queue.dequeue();
33
34// Destroy the queue when no longer needed
35queue.destroy();

StreamQueueManager

Overview

The StreamQueueManager class acts as a manager for multiple StreamQueue instances, allowing you to organize and control queues efficiently.

Usage

1import { StreamQueueManager, StreamQueueManagerOption } from '@hao-rxtool/queue';
2
3// Create a queue manager with optional options
4const options: StreamQueueManagerOption = { poolSize: 5 };
5const manager = new StreamQueueManager(options);
6
7
8const queueName = 'exampleQueue';
9// Listen for status changes of a specific queue
10manager.statusChange(queueName).subscribe((status) => {
11  console.log(`Queue status: ${status}`);
12});
13
14// Listen for an item is ready to be dequeued of a specific queue
15manager.whenReadyToDequeue(queueName).subscribe((item) => {
16  console.log(`When ready to dequeue item: ${item}`);
17});
18
19// Listen for an item has been dequeued of a specific queue
20manager.whenDequeued(queueName).subscribe((item) => {
21  console.log(`Dequeued item: ${item}`);
22});
23
24// Enqueue an item to the specific queue
25manager.enqueue(queueName, 1);
26
27// Dequeue an item from a queue
28manager.dequeue(queueName);
29
30// Enqueues an item to the specific queue and waits for it to be dequeued
31manager.enqueueAndWaitDequeue(queueName, 2).subscribe((item) => {
32  console.log(`Enqueue and wait dequeue item: ${item}`);
33});
34
35// Clear all queues from the manager
36manager.clear();

License

This library is licensed under the MIT License - see the LICENSE file for details.