Module: Cosmo::Batch::Dispatcher

Included in:
Cosmo::Batch
Defined in:
lib/cosmo/batch/dispatcher.rb,
sig/cosmo/batch/dispatcher.rbs

Overview

The engine behind Batch.notify: decides when a batch is done and fires its callbacks.

Instance Method Summary collapse

Instance Method Details

#dispatch_callback(callback_entry, bid, event, stats) ⇒ void

This method returns an undefined value.

Parameters:

  • callback_entry (Object)
  • bid (::String)
  • event (Symbol)
  • stats (Hash[Symbol, untyped]) 


79
80
81
82
83
 
# File 'lib/cosmo/batch/dispatcher.rb', line 79

def dispatch_callback(callback_entry, bid, event, stats)
  callback = Utils::Json.parse(callback_entry.value)
  status = stats.merge(bid: bid)
  Callback.perform_async(callback[:class], event, status, callback[:opts] || {})
end

#finalize(bid) ⇒ void

This method returns an undefined value.

Runs once per batch. Has its own "already done" guard, separate from try_fire's, because calling #jobs again on an already-finished batch would otherwise re-run this and double-report to the parent batch.

Parameters:

  • bid (::String) 


22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/cosmo/batch/dispatcher.rb', line 22

def finalize(bid)
  kv.create("#{bid}.finalized", "1")

  total = counter.get("#{bid}.total")
  failed = counter.get("#{bid}.failed")
  stats = { total: total, succeeded: total - failed, failed: failed }
  kv.set("#{bid}.ready", Utils::Json.dump(stats))
  purge_counters(bid)

  EVENTS.each { |event| try_fire(bid, event, stats: stats) }
  parent_propagate(bid, failed: failed)
rescue NATS::KeyValue::KeyWrongLastSequenceError
  # already finalized - this is a stray completion from a reused Batch, ignore it
end

#fireable?(callback_entry, event, stats) ⇒ Boolean

Parameters:

  • callback_entry (Object)
  • event (Symbol)
  • stats (Hash[Symbol, untyped], nil)

Returns:

  • (Boolean) 


72
73
74
75
76
77
 
# File 'lib/cosmo/batch/dispatcher.rb', line 72

def fireable?(callback_entry, event, stats)
  return false unless callback_entry && stats
  return false if event == :success && stats[:failed].to_i.positive?

  true
end

#link(parent_id) ⇒ void

This method returns an undefined value.

Adds this batch as one job of the parent batch.

Parameters:

  • parent_id (::String) 


53
54
55
56
 
# File 'lib/cosmo/batch/dispatcher.rb', line 53

def link(parent_id)
  counter.increment("#{parent_id}.total")
  counter.increment("#{parent_id}.pending")
end

#parent_propagate(bid, failed:) ⇒ void

This method returns an undefined value.

Tells the parent batch "one of your jobs (me) is done", counting any failure in this batch as one failure for the parent. Same shape as #notify - the child's bid just stands in for a jid here.

Parameters:

  • bid (::String)
  • failed: (::Integer) 


44
45
46
47
48
49
50
 
# File 'lib/cosmo/batch/dispatcher.rb', line 44

def parent_propagate(bid, failed:)
  meta = Utils::Json.parse(kv.get("#{bid}.meta")&.value)
  parent_id = meta && meta[:parent_id]
  return unless parent_id

  notify(parent_id, bid, success: failed.zero?)
end

#purge_counters(bid) ⇒ void

This method returns an undefined value.

Parameters:

  • bid (::String) 


37
38
39
 
# File 'lib/cosmo/batch/dispatcher.rb', line 37

def purge_counters(bid)
  %w[total pending failed].each { |key| counter.purge("#{bid}.#{key}") }
end

#release_pending(bid, msg_id: nil) ⇒ void

This method returns an undefined value.

Decrements pending. If this decrement just brought it to zero, finalize the batch. This never touches total or failed - that's what lets #jobs release its placeholder slot without messing up the stats we report back.

Parameters:

  • bid (::String)
  • msg_id: (::String, nil) (defaults to: nil) 


14
15
16
17
 
# File 'lib/cosmo/batch/dispatcher.rb', line 14

def release_pending(bid, msg_id: nil)
  pending = counter.decrement("#{bid}.pending", msg_id: msg_id)
  finalize(bid) if pending.to_i.zero?
end

#try_fire(bid, event, stats: nil) ⇒ void

This method returns an undefined value.

Fires the callback for event, if one is registered and the batch is ready. Safe to call any number of times from either #on or a job finishing - it only actually fires once.

Parameters:

  • bid (::String)
  • event (Symbol)
  • stats: (Hash[Symbol, untyped], nil) (defaults to: nil) 


61
62
63
64
65
66
67
68
69
70
 
# File 'lib/cosmo/batch/dispatcher.rb', line 61

def try_fire(bid, event, stats: nil)
  callback_entry = kv.get("#{bid}.callback.#{event}")
  stats ||= Utils::Json.parse(kv.get("#{bid}.ready")&.value)
  return unless fireable?(callback_entry, event, stats)

  kv.create("#{bid}.fired.#{event}", "1")
  dispatch_callback(callback_entry, bid, event, stats)
rescue NATS::KeyValue::KeyWrongLastSequenceError
  # someone else already fired this event, do nothing
end