123456789_123456789_123456789_123456789_123456789_

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

Instance Method Details

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

[ GitHub ] 

  


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



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 419



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


  








  

    #xackdel(key, group, *ids, keepref: nil, delref: nil, acked: nil)  ⇒ Array<Integer> 
  



  

    
Acknowledges and conditionally deletes one or more entries for a stream consumer
group, in a single atomic operation. XACKDEL combines XACK and XDELEX.

  





    

    
Examples:

        
With splatted entry ids, keeping consumer group references (the default)

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

        
With arrayed entry ids

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

        
Also drop dangling consumer group references

      
redis.xackdel('mystream', 'mygroup', '0-1', delref: true)

        
Only delete entries acknowledged by every consumer group

      
redis.xackdel('mystream', 'mygroup', '0-1', acked: true)

  


Parameters:


    
  
  • 
    key
    (String)
—    
    the stream key
    
  
    • 
  
  • 
    group
    (String)
—    
    the consumer group name
    
  
    • 
  
  • 
    ids
    (Array<String>)
—    
    one or multiple entry ids
    
  
    • 
  
  • 
    keepref
    (Boolean)
—    
    acknowledge and delete the entries but keep existing consumer
group PEL references (the server's default). Mutually exclusive with :delref and
:acked.
    
  
    • 
  
  • 
    delref
    (Boolean)
—    
    acknowledge and delete the entries and remove all consumer
group PEL references to them, including dangling ones. Mutually exclusive with
:keepref and :acked.
    
  
    • 
  
  • 
    acked
    (Boolean)
—    
    acknowledge the entries but only delete those already
acknowledged by every consumer group. Mutually exclusive with :keepref and
:delref.
    
  
    • 



Returns:


    
  
  • 
    (Array<Integer>)
—    
    one entry per requested id, in order: -1 if the key or that
id does not exist, 1 if the entry was acknowledged and deleted, 2 if it was
acknowledged but not deleted because references still exist (:acked)
    
  
    • 





  [ GitHub ]


  

  


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



def xackdel(key, group, *ids, keepref: nil, delref: nil, acked: nil)
  if [keepref, delref, acked].count { |option| option } > 1
    raise ArgumentError, "keepref, delref, and acked are mutually exclusive"
  end

  ids = ids.flatten
  args = [:xackdel, key, group]
  args << "KEEPREF" if keepref
  args << "DELREF" if delref
  args << "ACKED" if acked
  args.concat(["IDS", ids.size, *ids])

  send_command(args)
end


  








  

    #xadd(key, entry, approximate: nil, maxlen: nil, minid: nil, limit: nil, nomkstream: nil, id: '*', policy: nil, idmp: nil, idmpauto: nil)  ⇒ 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)

        
Idempotent production (Redis 8.6)

      
redis.xadd('mystream', { f1: 'v1' }, idmp: ['producer-1', 'msg-42'])
redis.xadd('mystream', { f1: 'v1' }, idmpauto: 'producer-1')

  


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 63



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

    pid, iid = idmp
    raise ArgumentError, "idmp must be [producer_id, idempotent_id]" if pid.nil? || iid.nil?

    args << 'IDMP' << pid << iid
  elsif idmpauto
    args << 'IDMPAUTO' << idmpauto
  end
  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 
  



  

    
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)
—    
    with the keys
    

      

    • 'next'    [String] the entry id to pass as start on the next call, 0-0 when done
      • 

    • 'entries' [Array<[String, Hash]>] the claimed entries as [id, fields] pairs, or just
the ids when justid is true
      • 

    • 'deleted' [Array] ids that were deleted from the stream and dropped from the
PEL (Redis 7.0); always [] on Redis 6.2
      • 

    
  
    • 





  [ GitHub ]


  

  


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



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


  








  

    #xcfgset(key, duration: nil, maxsize: nil)  ⇒ String 
  



  

    
Configures idempotent-producer (IDMP) tracking for an existing stream (Redis 8.6).
Changing either value clears the stream's tracked producer/idempotent-id pairs.

  





    

    
Examples:

      
redis.xcfgset('mystream', duration: 300, maxsize: 1000)

  


Parameters:


    
  
  • 
    key
    (String)
—    
    the stream key, must exist
    
  
    • 
  
  • 
    duration
    (Integer)
—    
    seconds an idempotent id is retained (1..86400, server default 100)
    
  
    • 
  
  • 
    maxsize
    (Integer)
—    
    most recent idempotent ids kept per producer (1..10000, server default 100)
    
  
    • 



Returns:


    
  
  • 
    (String)
—    
    OK
    
  
    • 



Raises:


    
  
  • 
    (ArgumentError)
  
    • 





  [ GitHub ]


  

  


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



def xcfgset(key, duration: nil, maxsize: nil)
  raise ArgumentError, "supply duration and/or maxsize" if duration.nil? && maxsize.nil?

  args = [:xcfgset, key]
  args << 'IDMP-DURATION' << Integer(duration) if duration
  args << 'IDMP-MAXSIZE' << Integer(maxsize) if maxsize
  send_command(args)
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 538



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 144



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


  








  

    #xdelex(key, *ids, keepref: nil, delref: nil, acked: nil)  ⇒ Array<Integer> 
  



  

    
Delete entries by entry ids, with control over consumer group PEL references. XDELEX
is an extension of XDEL.

  





    

    
Examples:

        
With splatted entry ids, keeping consumer group references (the default)

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

        
With arrayed entry ids

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

        
Also drop dangling consumer group references

      
redis.xdelex('mystream', '0-1', delref: true)

        
Only delete entries acknowledged by every consumer group

      
redis.xdelex('mystream', '0-1', acked: true)

  


Parameters:


    
  
  • 
    key
    (String)
—    
    the stream key
    
  
    • 
  
  • 
    ids
    (Array<String>)
—    
    one or multiple entry ids
    
  
    • 
  
  • 
    keepref
    (Boolean)
—    
    delete the entries but keep existing consumer group PEL
references (the server's default). Mutually exclusive with :delref and :acked.
    
  
    • 
  
  • 
    delref
    (Boolean)
—    
    delete the entries and remove all consumer group PEL references
to them, including dangling ones. Mutually exclusive with :keepref and :acked.
    
  
    • 
  
  • 
    acked
    (Boolean)
—    
    only delete entries already acknowledged by every consumer group.
Mutually exclusive with :keepref and :delref.
    
  
    • 



Returns:


    
  
  • 
    (Array<Integer>)
—    
    one entry per requested id, in order: -1 if the key or that
id does not exist, 1 if the entry was deleted, 2 if it was not deleted because
references still exist (:acked) or the stream has no consumer groups
    
  
    • 





  [ GitHub ]


  

  


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



def xdelex(key, *ids, keepref: nil, delref: nil, acked: nil)
  if [keepref, delref, acked].count { |option| option } > 1
    raise ArgumentError, "keepref, delref, and acked are mutually exclusive"
  end

  ids = ids.flatten
  args = [:xdelex, key]
  args << "KEEPREF" if keepref
  args << "DELREF" if delref
  args << "ACKED" if acked
  args.concat(["IDS", ids.size, *ids])

  send_command(args)
end


  








  

    #xgroup(subcommand, key, group, id_or_consumer = nil, mkstream: false, entriesread: nil)  ⇒ 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')

        
With create subcommand and the group's entries-read counter (Redis 7.0)

      
redis.xgroup(:create, 'mystream', 'mygroup', '$', mkstream: true, entriesread: 10)

  


Parameters:


    
  
  • 
    subcommand
    (String)
—    
    create setid destroy delconsumer createconsumer
    
  
    • 
  
  • 
    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 or createconsumer
      • 

    
  
    • 
  
  • 
    mkstream
    (Boolean)
—    
    (create) create an empty stream automatically if it does not exist
    
  
    • 
  
  • 
    entriesread
    (Integer)
—    
    (create, setid) the number of entries the group has read, used by
the server to compute the group's lag (Redis 7.0)
    
  
    • 



Returns:


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





  [ GitHub ]


  

  


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



def xgroup(subcommand, key, group, id_or_consumer = nil, mkstream: false, entriesread: nil)
  args = [:xgroup, subcommand, key, group, id_or_consumer].compact
  args << 'MKSTREAM' if mkstream
  args << 'ENTRIESREAD' << Integer(entriesread) if entriesread
  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 242



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 495



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 607



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 205



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 271



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, claim: nil, noack: nil)  ⇒ Hash{String => Array<Array>} 
  



  

    
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)

        
Also claim entries pending for over a minute on other consumers (Redis 8.4+)

      
redis.xreadgroup('mygroup', 'consumer1', 'mystream', '>', claim: 60_000)
# => { 'mystream' => [['0-1', { 'f' => 'v' }, 75_000, 2], ['0-2', { 'f' => 'v' }, 0, 0]] }

  


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 => Array<Array>})
—    
    the entries per stream, each [id, fields]
    
  
    • 





  [ GitHub ]


  

  


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



def xreadgroup(group, consumer, keys, ids, count: nil, max_count: nil, max_size: nil, block: nil, claim: 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 << 'CLAIM' << Integer(claim) if claim
  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 228



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


  








  

    #xsetid(key, id, entriesadded: nil, maxdeletedid: nil)  ⇒ String 
  



  

    
Set the last entry id of the stream, and optionally its entries-added counter and
maximal deleted entry id.

  





    

    
Examples:

        
Set the last id

      
redis.xsetid('mystream', '1526919030474-55')

        
Also set the counters used for lag tracking (Redis 7.0)

      
redis.xsetid('mystream', '1526919030474-55', entriesadded: 100, maxdeletedid: '1526919030474-10')

  


Parameters:


    
  
  • 
    key
    (String)
—    
    the stream key
    
  
    • 
  
  • 
    id
    (String)
—    
    the new last entry id, must not be smaller than the current one
    
  
    • 
  
  • 
    entriesadded
    (Integer)
—    
    the number of entries added to the stream over its lifetime (Redis 7.0)
    
  
    • 
  
  • 
    maxdeletedid
    (String)
—    
    the maximal entry id that was deleted from the stream (Redis 7.0)
    
  
    • 



Returns:


    
  
  • 
    (String)
—    
    OK
    
  
    • 





  [ GitHub ]


  

  


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



def xsetid(key, id, entriesadded: nil, maxdeletedid: nil)
  args = [:xsetid, key, id]
  args << 'ENTRIESADDED' << Integer(entriesadded) if entriesadded
  args << 'MAXDELETEDID' << maxdeletedid if maxdeletedid
  send_command(args)
end


  








  

    #xstream_ref_policy_token(policy)   (private)
  

  [ GitHub ]


  

  


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



def xstream_ref_policy_token(policy)
  token = policy.to_s.upcase
  unless %w[KEEPREF DELREF ACKED].include?(token)
    raise ArgumentError, "policy must be :keepref, :delref, or :acked"
  end

  token
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
      
  
      • 
  
    • 
    policy
    (Symbol, String)
—    
      (Redis 8.2) how trimming handles consumer group
PEL references to the evicted entries: :keepref (the server's default) keeps
them, :delref also removes them, :acked only evicts entries already
acknowledged by every consumer group
      
  
      • 

    


    

      
    • 
  


Returns:


    
  
  • 
    (Integer)
—    
    the number of entries actually deleted
    
  
    • 





  [ GitHub ]


  

  


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



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

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