Streams

Stream commands append entries to a log and read them back by ID or through consumer groups. These are the single-command forms. For continuous readers and acknowledged processing, use RedisStreams.

Example (Appending and Reading Entries)

import {  } from "effect";

import {  } from "@effect-kit/redis";

const  = .(function* () {
  const  = yield* ;

  const id = yield* .("orders", "*", [       
const id: `${bigint}-${bigint}`
    ["sku", "A-100"],
    ["qty", "2"],
  ]);

  const batch = yield* .([["orders", "0"]], { : 10 });        
const batch: Option<readonly StreamRead<StreamEntry>[]>

  return { ,  };
});

xadd returns the new entry ID as a ${bigint}-${bigint} template literal type. xread returns None when no stream has new entries.

Methods

Method	Command	Reply type	Purpose
`redis.xadd`	`XADD`	`${bigint}-${bigint}`	Append an entry and return its ID.
`redis.xaddIfExists`	`XADD`	Option<`${bigint}-${bigint}`>	Append an entry only if the stream exists (NOMKSTREAM).
`redis.xaddPayload`	`XADD`	`${bigint}-${bigint}`	Append a single-field entry encoded with a StreamPayload codec.
`redis.xlen`	`XLEN`	`bigint`	Return the number of entries.
`redis.xrange`	`XRANGE`	`readonly StreamEntry[]`	Read entries between two IDs.
`redis.xrevrange`	`XREVRANGE`	`readonly StreamEntry[]`	Read entries between two IDs, newest first.
`redis.xdel`	`XDEL`	`bigint`	Delete entries by ID.
`redis.xdelex`	`XDELEX`	`readonly (1n \| -1n \| 2n)[]`	Delete entries by ID with a reference policy.
`redis.xtrim`	`XTRIM`	`bigint`	Trim a stream by length or minimum ID.
`redis.xcfgset`	`XCFGSET`	`void`	Set stream idempotency options.
`redis.xread`	`XREAD`	`Option<readonly StreamRead<StreamEntry>[]>`	Read new entries without blocking.
`redis.xreadgroup`	`XREADGROUP`	`Option<readonly StreamRead<GroupEntry>[]>`	Read entries for a consumer group without blocking.
`redis.xreadgroupClaim`	`XREADGROUP`	`Option<readonly StreamRead<ClaimedGroupEntry>[]>`	Read for a consumer group and claim idle entries.
`redis.xack`	`XACK`	`bigint`	Acknowledge entries for a group.
`redis.xackdel`	`XACKDEL`	`readonly (1n \| -1n \| 2n)[]`	Acknowledge entries and delete them under a reference policy.
`redis.xnack`	`XNACK`	`bigint`	Negatively acknowledge entries for a group.
`redis.xgroupCreate`	`XGROUP CREATE`	`void`	Create a consumer group.
`redis.xgroupSetid`	`XGROUP SETID`	`void`	Move a group's last delivered ID.
`redis.xgroupDestroy`	`XGROUP DESTROY`	`boolean`	Delete a consumer group.
`redis.xgroupCreateconsumer`	`XGROUP CREATECONSUMER`	`boolean`	Create a consumer in a group.
`redis.xgroupDelconsumer`	`XGROUP DELCONSUMER`	`bigint`	Delete a consumer and return its pending count.
`redis.xgroupHelp`	`XGROUP HELP`	`readonly string[]`	Return the XGROUP help text.
`redis.xpending`	`XPENDING`	`PendingSummary`	Summarize a group's pending entries.
`redis.xpendingEntries`	`XPENDING`	`readonly PendingEntry[]`	List pending entries with owner and idle time.
`redis.xclaim`	`XCLAIM`	`readonly Option<GroupEntry>[]`	Claim pending entries for a consumer.
`redis.xclaimIds`	`XCLAIM`	readonly `${bigint}-${bigint}`[]	Claim pending entries and return only their IDs.
`redis.xautoclaim`	`XAUTOCLAIM`	`ClaimPage<Option<GroupEntry>>`	Claim idle pending entries from a start ID.
`redis.xautoclaimIds`	`XAUTOCLAIM`	ClaimPage<`${bigint}-${bigint}`>	Claim idle entries and return only their IDs.
`redis.xinfoStream`	`XINFO STREAM`	`StreamInfo`	Return stream statistics.
`redis.xinfoStreamFull`	`XINFO STREAM`	`StreamInfoFull`	Return full stream, group and consumer detail (FULL).
`redis.xinfoGroups`	`XINFO GROUPS`	`readonly StreamGroupInfo[]`	List consumer groups.
`redis.xinfoConsumers`	`XINFO CONSUMERS`	`readonly StreamConsumerInfo[]`	List a group's consumers.
`redis.xinfoHelp`	`XINFO HELP`	`readonly string[]`	Return the XINFO help text.

Blocking reads

Blocking commands need a session

redis.blpop and the other blocking commands do not exist on Redis. They hold a connection while they wait, so they live on RedisSessions and take BlockingOptions. See Sessions.

redis.xread and redis.xreadgroup accept StreamReadOptions without block. Reads that wait are on RedisSessions and take BlockingReadOptions or BlockingGroupOptions, which require block.

Method	Command	Reply type	Purpose
`sessions.xreadBlocking`	`XREAD`	`Option<readonly StreamRead<StreamEntry>[]>`	Read new entries, waiting up to the block time.
`sessions.xreadgroupBlocking`	`XREADGROUP`	`Option<readonly StreamRead<GroupEntry>[]>`	Read for a consumer group, waiting up to the block time.
`sessions.xreadgroupClaimBlocking`	`XREADGROUP`	`Option<readonly StreamRead<ClaimedGroupEntry>[]>`	Read for a group and claim idle entries, waiting up to the block time.

`StreamReadOptions`

xread takes ReadStreams, a non-empty tuple of [key, ReadId]. xreadgroup takes a group, a consumer and GroupReadStreams, a non-empty tuple of [key, GroupReadId]. ReadId is a StreamPosition, "$" or "+". GroupReadId is a StreamPosition or ">".

Prop

Type

GroupReadOptions has the same fields plus noAck?: boolean.

`StreamAddOptions`

xadd and xaddIfExists take an ID of the form "*", "<ms>-*" or "<ms>-<seq>" (AddId), a non-empty tuple of [field, value] pairs (StreamFieldInput), and StreamAddOptions.

Prop

Type

Group and claim options

GroupCreateOptions for xgroupCreate: { mkstream?: boolean; entriesRead?: number | bigint }. GroupSetIdOptions for xgroupSetid is the same without mkstream. GroupStart is a StreamPosition or "$".
PendingOptions for xpendingEntries: { start; end; count; idleMillis?; consumer? }.
ClaimOptions for xclaim: { deliveryTime?: { type: "IDLE" | "TIME"; millis }; retryCount?; force?; lastId? }.
AutoClaimOptions for xautoclaim: { count? }.
StreamNackOptions for xnack: { retryCount?; force? }. The mode argument is "SILENT" | "FAIL" | "FATAL".
StreamRangeOptions for xrange: { count? }. StreamConfigOptions for xcfgset: { idmpDurationSeconds?; idmpMaxSize? }.
StreamTrim for xtrim and ReferencePolicy as above. StreamIds is a non-empty tuple of StreamId.

Reply types

StreamEntry is { id: StreamId; fields: ReadonlyArray<readonly [Uint8Array, Uint8Array]> }.
GroupEntry is the same but fields is an Option, because a pending entry can outlive its payload.
ClaimedGroupEntry adds idleMillis and deliveries.
StreamRead<A> is { key: Uint8Array; entries: ReadonlyArray<A> }.
PendingSummary is { count; smallest; largest; consumers }. PendingEntry is { id; consumer; idleMillis; deliveries }.
ClaimPage<A> is { cursor; entries; deleted }.
StreamInfo, StreamInfoFull, StreamGroupInfo and StreamConsumerInfo are the XINFO replies.

MethodsCopy Anchor Link

Blocking readsCopy Anchor Link

StreamReadOptionsCopy Anchor Link

StreamAddOptionsCopy Anchor Link

Group and claim optionsCopy Anchor Link

Reply typesCopy Anchor Link