Class: Wurk::BatchSet

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/wurk/batch_set.rb

Overview

Discovery API over the global batches sorted set. size, each (yields Status), scan_tags(tag) for tag-indexed lookup.

Spec: docs/target/sidekiq-pro.md §2.7.

Direct Known Subclasses

Wurk::Batch::DeadSet

Constant Summary collapse

PAGE_SIZE = 
100
TAG_SCAN_LIMIT = 
20_000

Instance Method Summary collapse

Constructor Details

#initialize(key: 'batches') ⇒ BatchSet

Returns a new instance of BatchSet.



16
17
18
 
# File 'lib/wurk/batch_set.rb', line 16

def initialize(key: 'batches')
  @key = key
end

Instance Method Details

#each ⇒ Object

Newest-first iteration of every batch indexed in the set. Yields Wurk::Batch::Status instances — they HGETALL on construction so the caller pays one Redis round-trip per batch.



27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/wurk/batch_set.rb', line 27

def each
  return enum_for(:each) unless block_given?

  page = 0
  loop do
    start = page * PAGE_SIZE
    stop  = start + PAGE_SIZE - 1
    bids  = Wurk.redis { |conn| conn.call('ZRANGE', @key, start, stop, 'REV') }
    bids.each { |bid| yield Wurk::Batch::Status.new(bid) }
    break if bids.size < PAGE_SIZE

    page += 1
  end
end

#scan_tags(tag, &block) ⇒ Object

Yields each BID indexed under the given tag (set at tags:<tag>). No JSON parsing; cheap iteration. Caller wraps in Status.new(bid) if it needs full state.



74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/wurk/batch_set.rb', line 74

def scan_tags(tag, &block)
  return enum_for(:scan_tags, tag) unless block_given?

  cursor = '0'
  Wurk.redis do |conn|
    loop do
      cursor, members = conn.call('SSCAN', "tags:#{tag}", cursor, 'COUNT', PAGE_SIZE)
      members.each(&block)
      break if cursor == '0'
    end
  end
end

#size ⇒ Object 



20
21
22
 
# File 'lib/wurk/batch_set.rb', line 20

def size
  Wurk.redis { |conn| conn.call('ZCARD', @key) }.to_i
end

#snapshot(offset, count) ⇒ Array(Integer, Array<Hash>)

The set's size and the #each-ordered page of count batches from offset, each as its Status#summary: the page's ids in one round trip and every row's hash and set sizes in a second, where #each pays a round trip per batch from the top of the set down. Internal reader for the dashboard.

Returns:

  • (Array(Integer, Array<Hash>)) 


49
50
51
52
53
54
 
# File 'lib/wurk/batch_set.rb', line 49

def snapshot(offset, count)
  Wurk.redis do |conn|
    size, bids = page_ids(conn, offset, count)
    [size.to_i, summaries(conn, bids)]
  end
end

#tagged_snapshot(tag, offset, count) ⇒ Array(Integer, Array<Hash>)

#snapshot narrowed to the batches tagged tag and still indexed in this set: [how many, the page of summaries]. The tag index is a SET, so ordering it newest-first reads every member's score (one ZMSCORE) — bounded at TAG_SCAN_LIMIT members, past which the rest are not listed. Internal reader for the dashboard.

Returns:

  • (Array(Integer, Array<Hash>)) 


63
64
65
66
67
68
69
 
# File 'lib/wurk/batch_set.rb', line 63

def tagged_snapshot(tag, offset, count)
  bids = scan_tags(tag).first(TAG_SCAN_LIMIT)
  Wurk.redis do |conn|
    ranked = newest_first(bids, bids.empty? ? [] : conn.call('ZMSCORE', @key, *bids))
    [ranked.size, summaries(conn, ranked[offset, count] || [])]
  end
end