simple-in-memory-queue

easily create and consume in-memory queues, for nodejs and the browser

install

npm install simple-in-memory-queue

use

create a queue

import { createQueue, QueueOrder } from 'simple-in-memory-queue';

// fifo: first in, first out (default for most queues)
const queue = createQueue<string>({ order: QueueOrder.FIRST_IN_FIRST_OUT });

// lifo: last in, first out (stack behavior)
const stack = createQueue<string>({ order: QueueOrder.LAST_IN_FIRST_OUT });

push, peek, and pop

queue.push('a');
queue.push(['b', 'c']);
queue.push(['d', 'e']);

// peek to view items without removal
queue.peek();             // ['a']
queue.peek(2);            // ['a', 'b']
queue.peek(queue.length); // ['a', 'b', 'c', 'd', 'e']
queue.peek(2, queue.length); // ['c', 'd', 'e'] — slice from index 2 to end

// pop to remove and return items
queue.pop();              // ['a']
queue.pop();              // ['b']
queue.pop(2);             // ['c', 'd']

listen to events

queue.on.push.subscribe({ consumer: ({ items }) => console.log(items) });
queue.on.peek.subscribe({ consumer: ({ items }) => console.log(items) });
queue.on.pop.subscribe({ consumer: ({ items }) => console.log(items) });

use with consumers

common queue consumption patterns, ready to use

debounce consumer

waits for activity to stop before consumption. the consumer is called after gap.milliseconds of silence.

usecases

• wait for user to stop scroll, type, or resize before response
• collapse rapid events into one

import { createQueueWithDebounceConsumer } from 'simple-in-memory-queue';

const queue = createQueueWithDebounceConsumer<string>({
  gap: { milliseconds: 100 },
  consumer: ({ items }) => console.log(items),
});

batch consumer

collects items and flushes them in batches. the consumer is called when either:

• the oldest item has waited threshold.milliseconds
• the queue has threshold.size items

usecases

• batch events before send to analytics
• collect writes before flush to database

import { createQueueWithBatchConsumer } from 'simple-in-memory-queue';

const queue = createQueueWithBatchConsumer<string>({
  threshold: { milliseconds: 100, size: 5 },
  consumer: ({ items }) => console.log(items),
});

resilient remote consumer

processes items one at a time with automatic retry and failure recovery.

features

• immediate: calls consumer as soon as an item is available
• retry: retries failed items up to threshold.retry times, with delay.retry between attempts
• discard: drops items that exceed the retry threshold, calls on.failurePermanent
• pause: stops consumption when threshold.pause items fail in a row, calls on.pause
• non-block: processes other items while failed items wait for retry

usecases

• send requests to remote apis with automatic retry
• deliver webhooks with failure recovery

import { createQueueWithResilientRemoteConsumer } from 'simple-in-memory-queue';

const queue = createQueueWithResilientRemoteConsumer<string>({
  consumer: async ({ item }) => console.log(item),
  threshold: {
    concurrency: 1,
    retry: 3,
    pause: 5,
  },
  delay: {
    retry: 100,
    visibility: 500, // optional: wait before first attempt
  },
  on: {
    failureAttempt: ({ item, attempt, error }) => {
      console.log(`attempt ${attempt} failed`, item, error);
    },
    failurePermanent: ({ item, error }) => {
      console.error('permanently failed', item, error);
    },
    pause: ({ failures }) => {
      console.error('queue paused', failures);
    },
  },
});

api

Queue<T>

| method | description | |--------|-------------| | push(item \| items[]) | add one or more items | | peek(n?) | view top n items without removal (default: 1) | | peek(start, end) | view items from start to end index | | pop(n?) | remove and return top n items (default: 1) | | pop(start, end) | remove items from start to end index | | length | current number of items | | on.push | event stream for push events | | on.peek | event stream for peek events | | on.pop | event stream for pop events |

QueueOrder

| value | description | |-------|-------------| | FIRST_IN_FIRST_OUT | oldest items first (fifo) | | LAST_IN_FIRST_OUT | newest items first (lifo) |

createQueueWithDebounceConsumer

| parameter | description | |-----------|-------------| | gap.milliseconds | time to wait after last push before consume | | consumer | ({ items }) => void called with all queued items |

createQueueWithBatchConsumer

| parameter | description | |-----------|-------------| | threshold.milliseconds | max time an item can wait before flush | | threshold.size | max items before flush | | consumer | ({ items }) => void called with batch |

createQueueWithResilientRemoteConsumer

| parameter | description | |-----------|-------------| | threshold.concurrency | max parallel consumers (currently: 1) | | threshold.retry | max retry attempts per item | | threshold.pause | sequential failures before pause | | delay.retry | milliseconds between retry attempts | | delay.visibility | milliseconds before first attempt (optional) | | consumer | ({ item }) => Promise<void> called per item | | on.failureAttempt | ({ item, attempt, error }) => void on retry | | on.failurePermanent | ({ item, error }) => void when discarded | | on.pause | ({ failures }) => void when paused |