Class: Wurk::FlowSet

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

Overview

Discovery over the flows sorted set: size, and each newest-first yielding Wurk::Flow::Status. The counterpart of BatchSet, and a separate class rather than a key: on that one because the two yield different objects — a set that returns the wrong kind of status to save a constructor is a listing that renders nothing.

A flow is discoverable only through this index. Nothing scans flow:*: KEYS is refused outright and a SCAN over a production keyspace to build a page of 25 is the pathology the index exists to avoid. The set is trimmed on the same two-axis bound as batches (see Batch.trim_bounds), so it cannot outgrow what a dashboard can walk.

Constant Summary collapse

PAGE_SIZE = 
100

Instance Method Summary collapse

Constructor Details

#initialize(key: Wurk::Keys::FLOWS_SET) ⇒ FlowSet

Returns a new instance of FlowSet.



22
23
24
 
# File 'lib/wurk/flow_set.rb', line 22

def initialize(key: Wurk::Keys::FLOWS_SET)
  @key = key
end

Instance Method Details

#each ⇒ Object

Newest first, a page of ids per round trip. Each yielded status reads its own header — one round trip per flow, like BatchSet — and reads no node records unless the caller asks for them.

A fid the index still holds whose record has expired yields a status that answers exists? == false, rather than being skipped: the caller rendering the page is the one that knows whether a tombstone row is worth showing, and silently dropping members would make a page shorter than the count it asked for with no way to tell why.



39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/wurk/flow_set.rb', line 39

def each
  return enum_for(:each) unless block_given?

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

    page += 1
  end
end

#size ⇒ Object 



26
27
28
 
# File 'lib/wurk/flow_set.rb', line 26

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