Module: Redis::Commands::Streams

Relationships & Source Files
Extension / Inclusion / Inheritance Descendants
Included In: `::Redis`, `::Redis::Commands`, `::Redis::MultiConnection`, `::Redis::PipelinedConnection`
Defined in:	lib/redis/commands/streams.rb

Instance Method Summary

#xack(key, group, *ids) ⇒ Integer

Removes one or multiple entries from the pending entries list of a stream consumer group.
#xadd(key, entry, approximate: nil, maxlen: nil, minid: nil, limit: nil, nomkstream: nil, id: '*') ⇒ String

Add new entry to the stream.
#xautoclaim(key, group, consumer, min_idle_time, start, count: nil, justid: false) ⇒ Hash{String => Hash}, Array<String>

Transfers ownership of pending stream entries that match the specified criteria.
#xclaim(key, group, consumer, min_idle_time, *ids, **opts) ⇒ Hash{String => Hash}, Array<String>

Changes the ownership of a pending entry.
#xdel(key, *ids) ⇒ Integer

Delete entries by entry ids.
#xgroup(subcommand, key, group, id_or_consumer = nil, mkstream: false) ⇒ String, Integer

Manages the consumer group of the stream.
#xinfo(subcommand, key, group = nil) ⇒ Hash⁺

Returns the stream information each subcommand.
#xlen(key) ⇒ Integer

Returns the number of entries inside a stream.
#xnack(key, group, mode, *ids, retrycount: nil, force: nil) ⇒ Integer

Releases pending entries back to the consumer group's Pending Entries List without acknowledging them.
#xpending(key, group, *args, idle: nil) ⇒ Hash⁺

Fetches not acknowledging pending entries.
#xrange(key, start = '-', range_end = '+', count: nil) ⇒ Array<Array<String, Hash>>

Fetches entries of the stream in ascending order.
#xread(keys, ids, count: nil, max_count: nil, max_size: nil, block: nil) ⇒ Hash{String => Hash{String => Hash}}

Fetches entries from one or multiple streams.
#xreadgroup(group, consumer, keys, ids, count: nil, max_count: nil, max_size: nil, block: nil, noack: nil) ⇒ Hash{String => Hash{String => Hash}}

Fetches a subset of the entries from one or multiple streams related with the consumer group.
#xrevrange(key, range_end = '+', start = '-', count: nil) ⇒ Array<Array<String, Hash>>

Fetches entries of the stream in descending order.
#xtrim(key, maxlen, strategy: 'MAXLEN', approximate: true) ⇒ Integer

Trims older entries of the stream if needed.
#_xread(args, keys, ids, blocking_timeout_msec) private

Instance Method Details

#_xread(args, keys, ids, blocking_timeout_msec) (private)

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 451


def _xread(args, keys, ids, blocking_timeout_msec)
  keys = keys.is_a?(Array) ? keys : [keys]
  ids = ids.is_a?(Array) ? ids : [ids]
  args << 'STREAMS'
  args.concat(keys)
  args.concat(ids)

  if blocking_timeout_msec.nil?
    send_command(args, &HashifyStreams)
  elsif blocking_timeout_msec.to_f.zero?
    send_blocking_command(args, 0, &HashifyStreams)
  else
    send_blocking_command(args, blocking_timeout_msec.to_f / 1_000, &HashifyStreams)
  end
end

#xack(key, group, *ids) ⇒ `Integer`

Removes one or multiple entries from the pending entries list of a stream consumer group.

Examples:

With a entry id

redis.xack('mystream', 'mygroup', '1526569495631-0')

With splatted entry ids

redis.xack('mystream', 'mygroup', '0-1', '0-2')

With arrayed entry ids

redis.xack('mystream', 'mygroup', %w[0-1 0-2])

Parameters:

key (String) —
the stream key
group (String) —
the consumer group name
ids (Array<String>) —
one or multiple entry ids

Returns:

(Integer) —
the number of entries successfully acknowledged

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 293


def xack(key, group, *ids)
  args = [:xack, key, group].concat(ids.flatten)
  send_command(args)
end

#xadd(key, entry, approximate: nil, maxlen: nil, minid: nil, limit: nil, nomkstream: nil, id: '*') ⇒ `String`

Add new entry to the stream.

Examples:

Without options

redis.xadd('mystream', f1: 'v1', f2: 'v2')

With options

redis.xadd('mystream', { f1: 'v1', f2: 'v2' }, id: '0-0', maxlen: 1000, approximate: true, nomkstream: true)

Parameters:

key (String) —
the stream key
entry (Hash) —
one or multiple field-value pairs
opts (Hash) —
several options for XADD command

Returns:

(String) —
the entry id

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 51


def xadd(key, entry, approximate: nil, maxlen: nil, minid: nil, limit: nil, nomkstream: nil, id: '*')
  args = [:xadd, key]
  args << 'NOMKSTREAM' if nomkstream
  if maxlen
    raise ArgumentError, "can't supply both maxlen and minid" if minid

    args << "MAXLEN"
    args << "~" if approximate
    args << maxlen
  elsif minid
    args << "MINID"
    args << "~" if approximate
    args << minid
  end
  args.concat(['LIMIT', limit]) if limit
  args << id
  args.concat(entry.flatten)
  send_command(args)
end

#xautoclaim(key, group, consumer, min_idle_time, start, count: nil, justid: false) ⇒ `Hash`{`String` => `Hash`}, `Array`<`String`>

Transfers ownership of pending stream entries that match the specified criteria.

Examples:

Claim next pending message stuck > 5 minutes and mark as retry

redis.xautoclaim('mystream', 'mygroup', 'consumer1', 3600000, '0-0')

Claim 50 next pending messages stuck > 5 minutes and mark as retry

redis.xclaim('mystream', 'mygroup', 'consumer1', 3600000, '0-0', count: 50)

Claim next pending message stuck > 5 minutes and don't mark as retry

redis.xclaim('mystream', 'mygroup', 'consumer1', 3600000, '0-0', justid: true)

Claim next pending message after this id stuck > 5 minutes and mark as retry

redis.xautoclaim('mystream', 'mygroup', 'consumer1', 3600000, '1641321233-0')

Parameters:

key (String) —
the stream key
group (String) —
the consumer group name
consumer (String) —
the consumer name
min_idle_time (Integer) —
the number of milliseconds
start (String) —
entry id to start scanning from or 0-0 for everything
count (Integer) —
number of messages to claim (default 1)
justid (Boolean) —
whether to fetch just an array of entry ids or not. Does not increment retry count when true

Returns:

(Hash{String => Hash}) —
the entries successfully claimed
(Array<String>) —
the entry ids successfully claimed if justid option is true

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 402


def xautoclaim(key, group, consumer, min_idle_time, start, count: nil, justid: false)
  args = [:xautoclaim, key, group, consumer, min_idle_time, start]
  if count
    args << 'COUNT' << count.to_s
  end
  args << 'JUSTID' if justid
  blk = justid ? HashifyStreamAutoclaimJustId : HashifyStreamAutoclaim
  send_command(args, &blk)
end

#xclaim(key, group, consumer, min_idle_time, *ids, **opts) ⇒ `Hash`{`String` => `Hash`}, `Array`<`String`>

Changes the ownership of a pending entry

Examples:

With splatted entry ids

redis.xclaim('mystream', 'mygroup', 'consumer1', 3600000, '0-1', '0-2')

With arrayed entry ids

redis.xclaim('mystream', 'mygroup', 'consumer1', 3600000, %w[0-1 0-2])

With idle option

redis.xclaim('mystream', 'mygroup', 'consumer1', 3600000, %w[0-1 0-2], idle: 1000)

With time option

redis.xclaim('mystream', 'mygroup', 'consumer1', 3600000, %w[0-1 0-2], time: 1542866959000)

With retrycount option

redis.xclaim('mystream', 'mygroup', 'consumer1', 3600000, %w[0-1 0-2], retrycount: 10)

With force option

redis.xclaim('mystream', 'mygroup', 'consumer1', 3600000, %w[0-1 0-2], force: true)

With justid option

redis.xclaim('mystream', 'mygroup', 'consumer1', 3600000, %w[0-1 0-2], justid: true)

Parameters:

key (String) —
the stream key
group (String) —
the consumer group name
consumer (String) —
the consumer name
min_idle_time (Integer) —
the number of milliseconds
ids (Array<String>) —
one or multiple entry ids
opts (Hash) —
several options for XCLAIM command

Options Hash (**opts):

:idle (Integer) —
the number of milliseconds as last time it was delivered of the entry
:time (Integer) —
the number of milliseconds as a specific Unix Epoch time
:retrycount (Integer) —
the number of retry counter
:force (Boolean) —
whether to create the pending entry to the pending entries list or not
:justid (Boolean) —
whether to fetch just an array of entry ids or not

Returns:

(Hash{String => Hash}) —
the entries successfully claimed
(Array<String>) —
the entry ids successfully claimed if justid option is true

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 369


def xclaim(key, group, consumer, min_idle_time, *ids, **opts)
  args = [:xclaim, key, group, consumer, min_idle_time].concat(ids.flatten)
  args.concat(['IDLE',       opts[:idle].to_i])  if opts[:idle]
  args.concat(['TIME',       opts[:time].to_i])  if opts[:time]
  args.concat(['RETRYCOUNT', opts[:retrycount]]) if opts[:retrycount]
  args << 'FORCE'                                if opts[:force]
  args << 'JUSTID'                               if opts[:justid]
  blk = opts[:justid] ? Noop : HashifyStreamEntries
  send_command(args, &blk)
end

#xdel(key, *ids) ⇒ `Integer`

Delete entries by entry ids.

Examples:

With splatted entry ids

redis.xdel('mystream', '0-1', '0-2')

With arrayed entry ids

redis.xdel('mystream', ['0-1', '0-2'])

Parameters:

key (String) —
the stream key
ids (Array<String>) —
one or multiple entry ids

Returns:

(Integer) —
the number of entries actually deleted

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 115


def xdel(key, *ids)
  args = [:xdel, key].concat(ids.flatten)
  send_command(args)
end

#xgroup(subcommand, key, group, id_or_consumer = nil, mkstream: false) ⇒ `String`, `Integer`

Manages the consumer group of the stream.

Examples:

With create subcommand

redis.xgroup(:create, 'mystream', 'mygroup', '$')

With setid subcommand

redis.xgroup(:setid, 'mystream', 'mygroup', '$')

With destroy subcommand

redis.xgroup(:destroy, 'mystream', 'mygroup')

With delconsumer subcommand

redis.xgroup(:delconsumer, 'mystream', 'mygroup', 'consumer1')

Parameters:

subcommand (String) —
create setid destroy delconsumer
key (String) —
the stream key
group (String) —
the consumer group name
id_or_consumer (String) (defaults to: nil) —
- the entry id or $, required if subcommand is create or setid
- the consumer name, required if subcommand is delconsumer
mkstream (Boolean) —
whether to create an empty stream automatically or not

Returns:

(String) —
OK if subcommand is create or setid
(Integer) —
effected count if subcommand is destroy or delconsumer

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 233


def xgroup(subcommand, key, group, id_or_consumer = nil, mkstream: false)
  args = [:xgroup, subcommand, key, group, id_or_consumer, (mkstream ? 'MKSTREAM' : nil)].compact
  send_command(args)
end

#xinfo(subcommand, key, group = nil) ⇒ `Hash`⁺

Returns the stream information each subcommand.

Examples:

stream

redis.xinfo(:stream, 'mystream')

groups

redis.xinfo(:groups, 'mystream')

consumers

redis.xinfo(:consumers, 'mystream', 'mygroup')

Parameters:

subcommand (String) —
e.g. stream groups consumers
key (String) —
the stream key
group (String) (defaults to: nil) —
the consumer group name, required if subcommand is consumers

Returns:

(Hash) —
information of the stream if subcommand is stream
(Array<Hash>) —
information of the consumer groups if subcommand is groups
(Array<Hash>) —
information of the consumers if subcommand is consumers

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 22


def xinfo(subcommand, key, group = nil)
  args = [:xinfo, subcommand, key, group].compact
  block = case subcommand.to_s.downcase
  when 'stream'              then Hashify
  when 'groups', 'consumers' then proc { |r| r.map(&Hashify) }
  end

  send_command(args, &block)
end

#xlen(key) ⇒ `Integer`

Returns the number of entries inside a stream.

Examples:

With key

redis.xlen('mystream')

Parameters:

key (String) —
the stream key

Returns:

(Integer) —
the number of entries

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 174


def xlen(key)
  send_command([:xlen, key])
end

#xnack(key, group, mode, *ids, retrycount: nil, force: nil) ⇒ `Integer`

Releases pending entries back to the consumer group's Pending Entries List without acknowledging them. Released entries become unowned, are placed at the head of the PEL and are immediately available for re-delivery via #xreadgroup CLAIM, #xclaim or #xautoclaim, bypassing the usual idle-timeout delay.

Examples:

Let another consumer retry two entries

redis.xnack('mystream', 'mygroup', 'FAIL', '0-1', '0-2')

Graceful shutdown with arrayed entry ids

redis.xnack('mystream', 'mygroup', :silent, %w[0-1 0-2])

Mark a poison message as permanently failed

redis.xnack('mystream', 'mygroup', 'FATAL', '0-1')

Parameters:

key (String) —
the stream key
group (String) —
the consumer group name
mode (String, Symbol) —
how the entry's delivery counter is adjusted: SILENT decrements it by one (the delivery "didn't count"), FAIL keeps it unchanged, FATAL sets it to the maximum, marking the entry permanently failed
ids (Array<String>) —
one or multiple entry ids
retrycount (Integer) —
set the delivery counter to an explicit value, overriding the mode adjustment (internal, rarely needed)
force (Boolean) —
create unowned PEL entries for ids that are not currently pending; each must exist in the stream (internal, rarely needed)

Returns:

(Integer) —
the number of entries released back to the group PEL; ids that are not pending are ignored and not counted

Raises:

(ArgumentError)

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 326


def xnack(key, group, mode, *ids, retrycount: nil, force: nil)
  mode = mode.to_s.upcase
  raise ArgumentError, "mode must be SILENT, FAIL or FATAL" unless %w[SILENT FAIL FATAL].include?(mode)

  ids = ids.flatten
  args = [:xnack, key, group, mode, "IDS", ids.size].concat(ids)
  args << "RETRYCOUNT" << Integer(retrycount) if retrycount
  args << "FORCE" if force
  send_command(args)
end

#xpending(key, group, *args, idle: nil) ⇒ `Hash`⁺

Fetches not acknowledging pending entries

Examples:

With key and group

redis.xpending('mystream', 'mygroup')

With range options

redis.xpending('mystream', 'mygroup', '-', '+', 10)

With range and idle time options

redis.xpending('mystream', 'mygroup', '-', '+', 10, idle: 9000)

With range and consumer options

redis.xpending('mystream', 'mygroup', '-', '+', 10, 'consumer1')

Parameters:

key (String) —
the stream key
group (String) —
the consumer group name
start (String) —
start first entry id of range
end (String) —
end last entry id of range
count (Integer) —
count the number of entries as limit
consumer (String) —
the consumer name
opts (Hash) —
a customizable set of options

Returns:

(Hash) —
the summary of pending entries
(Array<Hash>) —
the pending entries details if options were specified

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 434


def xpending(key, group, *args, idle: nil)
  command_args = [:xpending, key, group]
  command_args << 'IDLE' << Integer(idle) if idle
  case args.size
  when 0, 3, 4
    command_args.concat(args)
  else
    raise ArgumentError, "wrong number of arguments (given #{args.size + 2}, expected 2, 5 or 6)"
  end

  summary_needed = args.empty?
  blk = summary_needed ? HashifyStreamPendings : HashifyStreamPendingDetails
  send_command(command_args, &blk)
end

#xrange(key, start = '-', range_end = '+', count: nil) ⇒ `Array`<`Array`<`String`, `Hash`>>

Fetches entries of the stream in ascending order.

Examples:

Without options

redis.xrange('mystream')

With a specific start

redis.xrange('mystream', '0-1')

With a specific start and end

redis.xrange('mystream', '0-1', '0-3')

With count options

redis.xrange('mystream', count: 10)

Parameters:

key (String) —
the stream key
start (String) (defaults to: '-') —
first entry id of range, default value is -
end (String) —
last entry id of range, default value is +
count (Integer) —
the number of entries as limit

Returns:

(Array<Array<String, Hash>>) —
the ids and entries pairs

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 137


def xrange(key, start = '-', range_end = '+', count: nil)
  args = [:xrange, key, start, range_end]
  args.concat(['COUNT', count]) if count
  send_command(args, &HashifyStreamEntries)
end

#xread(keys, ids, count: nil, max_count: nil, max_size: nil, block: nil) ⇒ `Hash`{`String` => `Hash`{`String` => `Hash`}}

Fetches entries from one or multiple streams. Optionally blocking.

Examples:

With a key

redis.xread('mystream', '0-0')

With multiple keys

redis.xread(%w[mystream1 mystream2], %w[0-0 0-0])

With count option

redis.xread('mystream', '0-0', count: 2)

With max_count option

redis.xread(%w[mystream1 mystream2], %w[0-0 0-0], max_count: 10)

With block option

redis.xread('mystream', '$', block: 1000)

Parameters:

keys (Array<String>) —
one or multiple stream keys
ids (Array<String>) —
one or multiple entry ids
count (Integer) —
the number of entries as limit per stream
max_count (Integer) —
the total number of entries as limit across all streams combined, must be greater than or equal to count when both are given (Redis 8.10+). Defaults to unlimited.
max_size (Integer) —
soft cap on the total reply size in bytes across all streams combined, measured on the serialized server reply including protocol overhead; a single oversized first entry can still be returned (Redis 8.10+). Defaults to unlimited.
block (Integer) —
the number of milliseconds as blocking timeout

Returns:

(Hash{String => Hash{String => Hash}}) —
the entries

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 203


def xread(keys, ids, count: nil, max_count: nil, max_size: nil, block: nil)
  args = [:xread]
  args << 'COUNT' << count if count
  args << 'MAXCOUNT' << Integer(max_count) if max_count
  args << 'MAXSIZE' << Integer(max_size) if max_size
  args << 'BLOCK' << block.to_i if block
  _xread(args, keys, ids, block)
end

#xreadgroup(group, consumer, keys, ids, count: nil, max_count: nil, max_size: nil, block: nil, noack: nil) ⇒ `Hash`{`String` => `Hash`{`String` => `Hash`}}

Fetches a subset of the entries from one or multiple streams related with the consumer group. Optionally blocking.

Examples:

With a key

redis.xreadgroup('mygroup', 'consumer1', 'mystream', '>')

With multiple keys

redis.xreadgroup('mygroup', 'consumer1', %w[mystream1 mystream2], %w[> >])

With count option

redis.xreadgroup('mygroup', 'consumer1', 'mystream', '>', count: 2)

With block option

redis.xreadgroup('mygroup', 'consumer1', 'mystream', '>', block: 1000)

With noack option

redis.xreadgroup('mygroup', 'consumer1', 'mystream', '>', noack: true)

Parameters:

group (String) —
the consumer group name
consumer (String) —
the consumer name
keys (Array<String>) —
one or multiple stream keys
ids (Array<String>) —
one or multiple entry ids
opts (Hash) —
several options for XREADGROUP command

Returns:

(Hash{String => Hash{String => Hash}}) —
the entries

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 269


def xreadgroup(group, consumer, keys, ids, count: nil, max_count: nil, max_size: nil, block: nil, noack: nil)
  args = [:xreadgroup, 'GROUP', group, consumer]
  args << 'COUNT' << count if count
  args << 'MAXCOUNT' << Integer(max_count) if max_count
  args << 'MAXSIZE' << Integer(max_size) if max_size
  args << 'BLOCK' << block.to_i if block
  args << 'NOACK' if noack
  _xread(args, keys, ids, block)
end

#xrevrange(key, range_end = '+', start = '-', count: nil) ⇒ `Array`<`Array`<`String`, `Hash`>>

Fetches entries of the stream in descending order.

Examples:

Without options

redis.xrevrange('mystream')

With a specific end

redis.xrevrange('mystream', '0-3')

With a specific end and start

redis.xrevrange('mystream', '0-3', '0-1')

With count options

redis.xrevrange('mystream', count: 10)

Parameters:

key (String) —
the stream key
end (String) —
first entry id of range, default value is +
start (String) (defaults to: '-') —
last entry id of range, default value is -

Returns:

(Array<Array<String, Hash>>) —
the ids and entries pairs

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 160


def xrevrange(key, range_end = '+', start = '-', count: nil)
  args = [:xrevrange, key, range_end, start]
  args.concat(['COUNT', count]) if count
  send_command(args, &HashifyStreamEntries)
end

#xtrim(key, maxlen, strategy: 'MAXLEN', approximate: true) ⇒ `Integer` #xtrim(key, minid, strategy: 'MINID', approximate: true) ⇒ `Integer`

Trims older entries of the stream if needed.

Examples:

Without options

redis.xtrim('mystream', 1000)

With options

redis.xtrim('mystream', 1000, approximate: true)

With strategy

redis.xtrim('mystream', '1-0', strategy: 'MINID')

Overloads:

#xtrim(key, maxlen, strategy: 'MAXLEN', approximate: true) ⇒ Integer
Parameters:
- key (String) —
  the stream key
- maxlen (Integer) —
  max length of entries
- strategy (String) —
  the limit strategy, must be MAXLEN
- approximate (Boolean) —
  whether to add ~ modifier of maxlen or not
- limit (Integer) —
  maximum count of entries to be evicted
#xtrim(key, minid, strategy: 'MINID', approximate: true) ⇒ Integer
Parameters:
- key (String) —
  the stream key
- minid (String) —
  minimum id of entries
- strategy (String) —
  the limit strategy, must be MINID
- approximate (Boolean) —
  whether to add ~ modifier of minid or not
- limit (Integer) —
  maximum count of entries to be evicted

Returns:

(Integer) —
the number of entries actually deleted

[ GitHub ]

# File 'lib/redis/commands/streams.rb', line 94


def xtrim(key, len_or_id, strategy: 'MAXLEN', approximate: false, limit: nil)
  strategy = strategy.to_s.upcase

  args = [:xtrim, key, strategy]
  args << '~' if approximate
  args << len_or_id
  args.concat(['LIMIT', limit]) if limit
  send_command(args)
end

Module: Redis::Commands::Streams

Instance Method Summary

Instance Method Details

#_xread(args, keys, ids, blocking_timeout_msec) (private)

#xack(key, group, *ids) ⇒ Integer

#xadd(key, entry, approximate: nil, maxlen: nil, minid: nil, limit: nil, nomkstream: nil, id: '*') ⇒ String

#xautoclaim(key, group, consumer, min_idle_time, start, count: nil, justid: false) ⇒ Hash{String => Hash}, Array<String>

#xclaim(key, group, consumer, min_idle_time, *ids, **opts) ⇒ Hash{String => Hash}, Array<String>

#xdel(key, *ids) ⇒ Integer

#xgroup(subcommand, key, group, id_or_consumer = nil, mkstream: false) ⇒ String, Integer

#xinfo(subcommand, key, group = nil) ⇒ Hash+

#xlen(key) ⇒ Integer

#xnack(key, group, mode, *ids, retrycount: nil, force: nil) ⇒ Integer

#xpending(key, group, *args, idle: nil) ⇒ Hash+

#xrange(key, start = '-', range_end = '+', count: nil) ⇒ Array<Array<String, Hash>>

#xread(keys, ids, count: nil, max_count: nil, max_size: nil, block: nil) ⇒ Hash{String => Hash{String => Hash}}

#xreadgroup(group, consumer, keys, ids, count: nil, max_count: nil, max_size: nil, block: nil, noack: nil) ⇒ Hash{String => Hash{String => Hash}}

#xrevrange(key, range_end = '+', start = '-', count: nil) ⇒ Array<Array<String, Hash>>

#xtrim(key, maxlen, strategy: 'MAXLEN', approximate: true) ⇒ Integer #xtrim(key, minid, strategy: 'MINID', approximate: true) ⇒ Integer

#xack(key, group, *ids) ⇒ `Integer`

#xadd(key, entry, approximate: nil, maxlen: nil, minid: nil, limit: nil, nomkstream: nil, id: '*') ⇒ `String`

#xautoclaim(key, group, consumer, min_idle_time, start, count: nil, justid: false) ⇒ `Hash`{`String` => `Hash`}, `Array`<`String`>

#xclaim(key, group, consumer, min_idle_time, *ids, **opts) ⇒ `Hash`{`String` => `Hash`}, `Array`<`String`>

#xdel(key, *ids) ⇒ `Integer`

#xgroup(subcommand, key, group, id_or_consumer = nil, mkstream: false) ⇒ `String`, `Integer`

#xinfo(subcommand, key, group = nil) ⇒ `Hash`⁺

#xlen(key) ⇒ `Integer`

#xnack(key, group, mode, *ids, retrycount: nil, force: nil) ⇒ `Integer`

#xpending(key, group, *args, idle: nil) ⇒ `Hash`⁺

#xrange(key, start = '-', range_end = '+', count: nil) ⇒ `Array`<`Array`<`String`, `Hash`>>

#xread(keys, ids, count: nil, max_count: nil, max_size: nil, block: nil) ⇒ `Hash`{`String` => `Hash`{`String` => `Hash`}}

#xreadgroup(group, consumer, keys, ids, count: nil, max_count: nil, max_size: nil, block: nil, noack: nil) ⇒ `Hash`{`String` => `Hash`{`String` => `Hash`}}

#xrevrange(key, range_end = '+', start = '-', count: nil) ⇒ `Array`<`Array`<`String`, `Hash`>>

#xtrim(key, maxlen, strategy: 'MAXLEN', approximate: true) ⇒ `Integer` #xtrim(key, minid, strategy: 'MINID', approximate: true) ⇒ `Integer`