Class: Wurk::Status::Progress

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/status/progress.rb

Overview

The in-job progress handle: status.at(row, total, 'importing').

Writes are coalesced. A job looping over 100k rows and reporting each one must not become 100k Redis round trips, so at most one write lands per INTERVAL; the newest unwritten values are buffered and the last thing the job said is always persisted — by #flush for a caller with nothing else to write, or by #drain for the server middleware, which folds the buffer into the terminal write it is making anyway. The first report is never delayed — a job that calls at once, early, is visible immediately.

Not thread-safe: one handle belongs to one job execution on one thread, the same contract IterableJob's cursor state has.

Constant Summary collapse

INTERVAL =

Coalescing window, seconds. Mirrors IterableJob::STATE_FLUSH_INTERVAL — a job checkpointing its own state and a job reporting progress are the same traffic problem, and two different answers to it would just be two numbers to tune.

5

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(jid, ttl: nil, interval: INTERVAL, pool: nil) ⇒ Progress

ttl: nil defers to status_ttl at write time rather than freezing the configured value into every handle the process builds.



28
29
30
31
32
33
34
35
 
# File 'lib/wurk/status/progress.rb', line 28

def initialize(jid, ttl: nil, interval: INTERVAL, pool: nil)
  @jid      = jid.to_s
  @ttl      = ttl
  @interval = interval
  @pool     = pool
  @pending  = {}
  @wrote_at = nil
end

Instance Attribute Details

#jid ⇒ Object (readonly)

Returns the value of attribute jid.



24
25
26
 
# File 'lib/wurk/status/progress.rb', line 24

def jid
  @jid
end

Instance Method Details

#at(num, total = nil, message = nil) ⇒ Integer

Report position. total and message are optional and sticky: pass them once and later bare at(n) calls keep them.

Returns:

  • (Integer) —

    num, so at can wrap an existing counter expression



41
42
43
44
45
46
47
 
# File 'lib/wurk/status/progress.rb', line 41

def at(num, total = nil, message = nil)
  @pending[:progress] = num.to_i
  @pending[:total]    = total.to_i unless total.nil?
  @pending[:message]  = message.to_s unless message.nil?
  maybe_write
  num
end

#drain ⇒ Hash

Hand the buffer back instead of writing it, and clear it. The caller takes over responsibility for persisting the values — ::Status does this so a job's last reported position rides along on the terminal complete/failed write rather than costing a second round trip of its own.

Returns:

  • (Hash) —

    the buffered fields, empty when nothing is pending



72
73
74
75
76
 
# File 'lib/wurk/status/progress.rb', line 72

def drain
  pending = @pending
  @pending = {}
  pending
end

#flush ⇒ Boolean

Force out whatever at/message buffered since the last write.

Returns:

  • (Boolean) —

    true when something was written



59
60
61
62
63
 
# File 'lib/wurk/status/progress.rb', line 59

def flush
  return false if @pending.empty?

  write
end

#message(text) ⇒ Object

Report a human-readable step without moving the counter.



50
51
52
53
54
 
# File 'lib/wurk/status/progress.rb', line 50

def message(text)
  @pending[:message] = text.to_s
  maybe_write
  text
end