module Redis::Commands::Stream

Direct including types

Defined in:

commands/stream.cr

Instance Method Summary

Instance Method Detail

def xack(key : String, group : String, id : String) #

[View source]
def xack(key : String, group : String, ids : Enumerable(String)) #

[View source]
def xackdel(key : String, group : String, delete_mode : DeleteMode | Nil, ids : Enumerable(String)) #

[View source]
def xackdel(key : String, group : String, ids : Enumerable(String)) #

[View source]
def xadd(key : String, id : String, fields : NamedTuple | ::Hash(String, String), *, idmpauto : String | Nil = nil, idmp : Tuple(String, String) | Nil = nil) #

Append an entry with the specified data to the stream with the given key and gives it the specified id. If the id is "*", Redis will assign it an id of the form "#{Time.utc.to_unix_ms}-#{autoincrementing_index}". If maxlen is provided, Redis will trim the stream to the specified length. If maxlen is of the form ~ 1000, Redis will trim it to approximately that length, removing entries when it can do so efficiently. This method returns the id that Redis stores.

redis.xadd "my-stream", "*", {"name" => "foo", "id" => UUID.random.to_s}

[View source]
def xadd(key : String, id : String, *, maxlen, fields : NamedTuple | ::Hash(String, String), idmpauto : String | Nil = nil, idmp : Tuple(String, String) | Nil = nil) #

Append an entry with the specified data to the stream with the given key and gives it the specified id. If the id is "*", Redis will assign it an id of the form "#{Time.utc.to_unix_ms}-#{autoincrementing_index}". If maxlen is provided, Redis will trim the stream to the specified length. If maxlen is of the form ~ 1000, Redis will trim it to approximately that length, removing entries when it can do so efficiently. This method returns the id that Redis stores.

redis.xadd "my-stream", "*", maxlen: {"~", "1000"}, fields: {"name" => "foo", "id" => UUID.random.to_s}

[View source]
def xadd(key : String, id : String, *, minid, fields : NamedTuple | ::Hash(String, String), idmpauto : String | Nil = nil, idmp : Tuple(String, String) | Nil = nil) #

[View source]
def xadd(key : String, id : String, **fields : String) #

Append an entry with the specified data to the stream with the given key and gives it the specified id. If the id is "*", Redis will assign it an id of the form "#{Time.utc.to_unix_ms}-#{autoincrementing_index}". If maxlen is provided, Redis will trim the stream to the specified length. If maxlen is of the form ~ 1000, Redis will trim it to approximately that length, removing entries when it can do so efficiently. This method returns the id that Redis stores.

redis.xadd "my-stream", "*", {name: "foo", id: UUID.random.to_s}

DEPRECATED Using keyword arguments for stream event fields is deprecated and will be removed in a future release. It causes conflicts with optional maxlen and minid when passed as string values. Use the fields: {foo: "bar"} overload instead.


[View source]
def xadd(key : String, id : String, *, maxlen, **fields : String) #

DEPRECATED Using keyword arguments for stream event fields is deprecated and will be removed in a future release. It causes conflicts with optional maxlen and minid when passed as string values. Use the fields: {foo: "bar"} overload instead.


[View source]
def xadd(key : String, id : String, *, minid, **fields : String) #

DEPRECATED Using keyword arguments for stream event fields is deprecated and will be removed in a future release. It causes conflicts with optional maxlen and minid when passed as string values. Use the fields: {foo: "bar"} overload instead.


[View source]
def xautoclaim(key : String, group : String, consumer : String, min_idle_time : Time::Span, start : String, count : Int32 | String | Nil = nil) #

[View source]
def xdel(key : String, ids : Enumerable(String)) #

[View source]
def xdel(key : String, *ids : String) #

[View source]
def xgroup(command : String, key : String, groupname : String) #

Run a Redis XGROUP subcommand for a given stream. See the XGROUP command in the Redis documentation for more information.

redis.xgroup "DESTROY", "my-stream", "my-group"

[View source]
def xgroup(command : XGroup, key : String, groupname : String, *, id : String | Nil = nil, mkstream = false, consumer_name : String | Nil = nil) #

Run a Redis XGROUP subcommand for a given stream. See the XGROUP command in the Redis documentation for more information.

redis.xgroup :create, "my-stream", "my-group", mkstream: true

[View source]
def xgroup(command : String, key : String, groupname : String, *args : String) #

Run a Redis XGROUP subcommand for a given stream. See the XGROUP command in the Redis documentation for more information.

redis.xgroup "CREATE", "my-stream", "my-group", "0"

[View source]
def xgroup_create(key : String, groupname : String, *, id : String = "$", mkstream = false) #

Create the consumer group groupname in the stream contained in key.


[View source]
def xgroup_create_consumer(key : String, groupname : String, consumer_name : String) #

Create a consumer consumer_name in the consumer group groupname in the stream contained in key.

consumer_id = UUID.v7.to_s
redis.xgroup_create "orders", "fulfillment", mkstream: true
redis.xgroup_create_consumer "orders", "fulfillment", consumer_id

[View source]
def xgroup_del_consumer(key : String, group : String, consumer : String) #

Delete the given consumer from the given group in the stream stored in key.


[View source]
def xgroup_destroy(key : String, group : String) #

Delete the consumer group group in the stream contained in key.


[View source]
def xinfo_consumers(key : String, group : String) #

[View source]
def xinfo_groups(key : String) #

[View source]
def xinfo_stream(key : String) #

Return the details about the stream stored in key.

stream = Redis::Streaming::XInfoStreamResponse.new(
  redis.xinfo_stream("orders")
)
# => Redis::Streaming::XInfoStreamResponse(
#     @entries_added=1,
#     @first_entry=
#      Redis::Streaming::Message(
#       @delivery_count=0,
#       @id="1780361273088-0",
#       @last_delivered_at=1970-01-01 00:00:00Z,
#       @values={"id" => "0"}),
#     @groups=1,
#     @idmp_duration=100,
#     @idmp_maxsize=100,
#     @iids_added=0,
#     @iids_duplicates=0,
#     @iids_tracked=0,
#     @last_entry=
#      Redis::Streaming::Message(
#       @delivery_count=0,
#       @id="1780361273088-0",
#       @last_delivered_at=1970-01-01 00:00:00Z,
#       @values={"id" => "0"}),
#     @last_generated_id="1780361273088-0",
#     @length=1,
#     @max_deleted_entry_id="0-0",
#     @pids_tracked=0,
#     @radix_tree_keys=1,
#     @radix_tree_nodes=2,
#     @recorded_first_entry_id="1780361273088-0")

[View source]
def xinfo_stream_full(key : String, *, count : Int | String | Nil = nil) #

[View source]
def xlen(key : String) #

Return the number of entries in the given stream


[View source]
def xnack(key : String, group : String, mode : NackMode, ids : Enumerable(String)) #

[View source]
def xpending(key : String, group : String, start : String, end finish : String, count : String | Int32, idle : String | Time::Span | Nil = nil) #

[View source]
def xpending(key : String, group : String) #

XPENDING key group [[IDLE min-idle-time] start end count [consumer]]


[View source]
def xrange(key : String, start min : String, end max : String, count : String | Int32 | Nil = nil) #

Return the entries in the given stream between the start and end ids. If count is provided, Redis will return only that number of entries.


[View source]
def xread(*, count : Int | String | Nil = nil, block : Time::Span | Int | String | Nil = nil, streams : NamedTuple) #

[View source]
def xread(*, count : Int | String | Nil = nil, block : Time::Span | Int | String | Nil = nil, streams : ::Hash(String, String)) #

[View source]
def xreadgroup(group : String, consumer : String, count : String | Int32 | Nil = nil, block : Time::Span | String | Int32 | Nil = nil, claim : Time::Span | String | Int32 | Nil = nil, no_ack = false, streams : ::Hash(String, String) = {} of String => String) #

Execute an XREADGROUP command on the Redis server.

This is returned in its raw form from Redis, but you can pass it to a Redis::Streaming::XReadGroupResponse to make it easier to work with.


[View source]
def xreadgroup(group : String, consumer : String, count : String | Int32 | Nil = nil, block : Time::Span | String | Int32 | Nil = nil, claim : Time::Span | String | Int32 | Nil = nil, no_ack = false, streams : NamedTuple = NamedTuple.new) #

Execute an XREADGROUP command on the Redis server. If block is not nil, the server will block for up to that much time (if you pass a number, it will be interpreted as milliseconds) until any new messages enter the stream.

This is returned in its raw form from Redis, but you can pass it to a Redis::Streaming::XReadGroupResponse to make it easier to work with.

# Long-poll for up to 10 messages from the stream with key `my_stream`,
# blocking for up to 2 seconds if there are no messages waiting.
response = redis.xreadgroup "group", "consumer",
  streams: {my_stream: ">"},
  count: 10,
  block: 2.seconds
response = Redis::Streaming::XReadGroupResponse.new(response)

[View source]
def xrevrange(key : String, end max : String, start min : String, count : String | Int32 | Nil = nil) #

Return the entries in the given stream between the start and end ids. If count is provided, Redis will return only that number of entries.


[View source]
def xtrim(key : String, *, maxlen : Tuple(String, String), limit : String | Int32 | Nil = nil, delete_mode : DeleteMode | Nil = nil) #

[View source]
def xtrim(key : String, *, minid : Tuple(String, String), limit : String | Int32 | Nil = nil, delete_mode : DeleteMode | Nil = nil) #

[View source]