Redis.Commands.Stream (Redis v0.8.0)

Copy Markdown View Source

Command builders for Redis stream operations.

This module provides pure functions that build Redis Stream command lists. Streams are append-only log structures that support fan-out reads, consumer groups with at-least-once delivery, and automatic ID generation. They are well-suited for event sourcing, task queues, and activity feeds.

Every function returns a plain list of strings (a command). To execute a command, pass the result to Redis.command/2; to batch several commands in a single round trip, use Redis.pipeline/2.

Response shapes

These functions build commands; they do not normalize Redis replies. Connections negotiate RESP3 by default and fall back to RESP2 when needed. In particular, XREAD and XREADGROUP return a map keyed by stream under RESP3:

{:ok, %{"events" => [["1234567890123-0", ["type", "click"]]]}}

The same reply under RESP2 is a list of stream/entries pairs:

{:ok, [["events", [["1234567890123-0", ["type", "click"]]]]]}

Both protocols return nil when no stream can be served. XRANGE and XREVRANGE return a list of entries under both protocols.

Examples

Appending entries and reading them back:

iex> Redis.command(conn, Redis.Commands.Stream.xadd("events", "*", [{"type", "click"}, {"url", "/home"}]))
{:ok, "1234567890123-0"}

iex> Redis.command(conn, Redis.Commands.Stream.xread(streams: [{"events", "0"}], count: 10))
{:ok, %{"events" => [["1234567890123-0", ["type", "click", "url", "/home"]]]}}

Consumer group pattern -- read, process, acknowledge:

iex> Redis.command(conn, Redis.Commands.Stream.xgroup_create("events", "workers", "0", mkstream: true))
{:ok, "OK"}

iex> Redis.command(conn, Redis.Commands.Stream.xreadgroup("workers", "worker-1", streams: [{"events", ">"}], count: 5))
{:ok, %{"events" => [["1234567890123-0", ["type", "click", "url", "/home"]]]}}

iex> Redis.command(conn, Redis.Commands.Stream.xack("events", "workers", ["1234567890123-0"]))
{:ok, 1}

Summary

Functions

xack(key, group, ids)

@spec xack(String.t(), String.t(), [String.t()]) :: [String.t()]

Builds an XACK command to acknowledge one or more stream entries.

Acknowledging an entry removes it from the consumer group's pending entries list (PEL). Returns the number of entries successfully acknowledged.

xadd(key, id \\ "*", fields, opts \\ [])

@spec xadd(String.t(), String.t(), [{String.t(), String.t()}], keyword()) :: [
  String.t()
]

Builds an XADD command to append an entry to the stream at key.

id defaults to "*" which lets Redis auto-generate a monotonic ID. fields is a list of {field, value} tuples representing the entry payload. Options:

  • :maxlen - cap the stream length with approximate trimming (~)

xautoclaim(key, group, consumer, min_idle_time, start, opts \\ [])

@spec xautoclaim(String.t(), String.t(), String.t(), integer(), String.t(), keyword()) ::
  [String.t()]

xclaim(key, group, consumer, min_idle_time, ids, opts \\ [])

@spec xclaim(String.t(), String.t(), String.t(), integer(), [String.t()], keyword()) ::
  [String.t()]

xdel(key, ids)

@spec xdel(String.t(), [String.t()]) :: [String.t()]

xgroup_create(key, group, id \\ "$", opts \\ [])

@spec xgroup_create(String.t(), String.t(), String.t(), keyword()) :: [String.t()]

xgroup_createconsumer(key, group, consumer)

@spec xgroup_createconsumer(String.t(), String.t(), String.t()) :: [String.t()]

xgroup_delconsumer(key, group, consumer)

@spec xgroup_delconsumer(String.t(), String.t(), String.t()) :: [String.t()]

xgroup_destroy(key, group)

@spec xgroup_destroy(String.t(), String.t()) :: [String.t()]

xgroup_setid(key, group, id)

@spec xgroup_setid(String.t(), String.t(), String.t()) :: [String.t()]

xinfo_consumers(key, group)

@spec xinfo_consumers(String.t(), String.t()) :: [String.t()]

xinfo_groups(key)

@spec xinfo_groups(String.t()) :: [String.t()]

xinfo_stream(key, opts \\ [])

@spec xinfo_stream(
  String.t(),
  keyword()
) :: [String.t()]

Builds an XINFO STREAM command to return metadata about the stream at key.

Pass full: true to include detailed information about every consumer group, consumer, and pending entry.

xlen(key)

@spec xlen(String.t()) :: [String.t()]

Builds an XLEN command to return the number of entries in the stream at key.

xpending(key, group, opts \\ [])

@spec xpending(String.t(), String.t(), keyword()) :: [String.t()]

xrange(key, start_id \\ "-", end_id \\ "+", opts \\ [])

@spec xrange(String.t(), String.t(), String.t(), keyword()) :: [String.t()]

xread(opts)

@spec xread(keyword()) :: [String.t()]

Builds an XREAD command to read entries from one or more streams.

Options:

  • :streams (required) - a list of {stream_key, last_id} tuples. Use "0" to read from the beginning or "$" to read only new entries.
  • :count - maximum number of entries to return per stream
  • :block - block for up to this many milliseconds waiting for new data

Execution returns a stream-keyed map under RESP3 and stream/entries pairs under RESP2. See "Response shapes" in the module documentation.

xreadgroup(group, consumer, opts)

@spec xreadgroup(String.t(), String.t(), keyword()) :: [String.t()]

Builds an XREADGROUP command to read entries via a consumer group.

Each entry delivered to consumer within group must later be acknowledged with xack/3. Options:

  • :streams (required) - a list of {stream_key, id} tuples. Use ">" as the id to receive only new, undelivered messages.
  • :count - maximum number of entries to return per stream
  • :block - block for up to this many milliseconds waiting for new data
  • :noack - do not require acknowledgement for delivered entries

Execution returns a stream-keyed map under RESP3 and stream/entries pairs under RESP2. See "Response shapes" in the module documentation.

xrevrange(key, end_id \\ "+", start_id \\ "-", opts \\ [])

@spec xrevrange(String.t(), String.t(), String.t(), keyword()) :: [String.t()]

xsetid(key, last_id, opts \\ [])

@spec xsetid(String.t(), String.t(), keyword()) :: [String.t()]

xtrim(key, opts \\ [])

@spec xtrim(
  String.t(),
  keyword()
) :: [String.t()]