redis-x-stream

library4.1.0jsnpmunverified

An async iterable interface for Redis streams, built on top of ioredis and msgpackr. Version 4.1.0 supports Redis 5+ (6.2+ for claimIdleTime). Provides high-level modules for job queues with workers, retries, backoff, dead-letter queues (import from 'redis-x-stream/queue') and a distributed single-flight cache with two-tier storage (import from 'redis-x-stream/cache'). Features typed parsing, consumer group support with automatic PEL re-delivery, dynamic stream addition, dead consumer recovery via XAUTOCLAIM, flush timers for pending acks, and observability methods (info, groups, consumers, pending). Ships TypeScript declarations. Releases follow semver with monthly cadence.

messaging database

npm install redis-x-stream

github ↗package ↗homepage ↗

INSTALL

IMPORT

SIG · REDIS-X-STREAM

redis-x-stream

messagingjavascriptv4.1.0

harness data pending

Install & Compatibility

Where this runs

No compatibility data collected yet for this library.

Code

Verified usage

Verified import paths — ran on the pinned version, not inferred.

RedisStream

✓ import { RedisStream } from 'redis-x-stream'

✗ const RedisStream = require('redis-x-stream')

ESM-only; CommonJS require will not work. Package only ships ESM.

Queue, Worker

✓ import { Queue, Worker } from 'redis-x-stream/queue'

✗ import { Queue } from 'redis-x-stream'

Queue and Worker are in a submodule; importing from the main entry will result in undefined.

SingleFlightCache

✓ import { SingleFlightCache } from 'redis-x-stream/cache'

✗ import { SingleFlightCache } from 'redis-x-stream'

Cache module is separate; no default export in main entry.

Creates a RedisStream consumer on 'events', iterates entries with ack-on-iterate, logs parsed key-value pairs, and gracefully quits.

import { RedisStream } from 'redis-x-stream'; import Redis from 'ioredis'; const redis = new Redis({ host: process.env.REDIS_HOST ?? 'localhost' }); async function main() { const stream = new RedisStream({ streams: ['events'], redisClient: redis, ackOnIterate: true, block: 5000, }); for await (const [name, [id, keyvals]] of stream) { console.log(`Stream: ${name}, ID: ${id}, Data:`, Object.fromEntries( keyvals.reduce((acc, val, i, arr) => i % 2 === 0 ? [...acc, [val]] : (acc[acc.length-1].push(val), acc), []) )); } await stream.quit(); redis.disconnect(); } main().catch(console.error);

Debug

Known issues

breakingIn v4, the main export has changed from a default export to a named export 'RedisStream'. Importing as default will break.

fix

Change `import RedisStream from 'redis-x-stream'` to `import { RedisStream } from 'redis-x-stream'`.

affects: >=4.0.0

breakingIn v4, the constructor options 'deleteAfterAck' has been renamed to 'deleteOnAck'.

fix

Replace `deleteAfterAck: true` with `deleteOnAck: true`.

affects: >=4.0.0

deprecatedThe method `close()` is deprecated in v4. Use `drain()` or `quit()` instead.

fix

Replace `stream.close()` with either `await stream.drain()` (graceful) or `await stream.quit()` (immediate).

affects: >=4.0.0 <5.0.0

gotchaThe 'block' option defaults to 0 (non-blocking) for consumer groups, but to 5000 (5 seconds) for standalone consumers. This mismatch can cause unexpected behavior in groups.

fix

Always explicitly set `block: Infinity` or a desired value when using consumer groups.

affects: >=4.0.0

gotchaWhen using `parse` callback with TypeScript, the generic type parameter must be specified explicitly; TypeScript may not infer the return type correctly.

fix

Use `new RedisStream<MyType>({ ... parse: (id, kv) => ({...}) })` and ensure the parse function returns the exact type.

affects: >=4.0.0

breakingIn v4, the library now uses ESM only and no longer supports CommonJS. Requires 'type': 'module' or .mjs extension.

fix

Add `"type": "module"` to package.json or rename files to .mjs. Use dynamic import() if needed.

affects: >=4.0.0

Errors

Common errors & fixes

ERR Wrong number of arguments for XREADGROUP command

The 'stream' option is misspelled as 'streams' (plural) is required and must be an array, but may be passed as a string.

fix

Ensure `streams: ['my-stream']` is an array, not a single string. Use `streams: ['my-stream']` instead of `stream: 'my-stream'`.

TypeError: redisClient.xread is not a function

ioredis instance is not passed or is not a valid ioredis client (e.g., passed null or wrong constructor).

fix

Pass a valid ioredis instance via `redisClient` option: `new RedisStream({ streams: ['s'], redisClient: new Redis() })`.

Message: Cannot find module 'redis-x-stream'

The package is installed incorrectly or not installed, or Node.js version <16.

fix

Run `npm install redis-x-stream ioredis` and ensure Node.js version >=16 (ESM support).

Upgrade

Version history

4.1.0latest on npm

Audit

Dependencies

ioredisrequiredPeer dependency: Redis client for all stream operations.

msgpackroptionalPeer dependency: serialization for queue and cache submodules.

Agent activity

27 hits · last 30 days

node

OpenAI (training)

Resources

githubhttps://github.com/calebboyd/redis-x-stream ↗

packageredis-x-stream ↗

homepagehttps://github.com/calebboyd/redis-x-stream#readme ↗