Redis Streams for Nodejs

redis-stream.ts

import { Readable, Writable } from "stream";
import { commandOptions, createClient } from "redis";

export const createWriteStream = (client: ReturnType<typeof createClient>, key: string, opts?: { ttlMs?: number }) =>
	new Writable({
		objectMode: false,
		construct: (callback) =>
			client
				.del(key)
				.then(() => callback())
				.catch(callback),

		write: (chunk, _, callback) =>
			client
				.append(key, chunk)
				.then(() => callback())
				.catch(callback),

		final: (callback) =>
			opts?.ttlMs
				? client
					.pExpire(key, opts?.ttlMs)
					.then(() => callback())
					.catch(callback)
				: callback(),
	});

export const createReadStream = (client: ReturnType<typeof createClient>, key: string) => {
	let sent = 0;
	return new Readable({
		objectMode: false,
		async read(size) {
			const data = await client.getRange(
				commandOptions({ returnBuffers: true }),
				key,
				sent,
				sent + size,
			);
			if (data.length === 0) {
				this.push(null);
				return;
			}
			sent += data.length;
			this.push(data);
		},
	});
};