Module: Pgbus::ActiveJob::BatchId

Extended by:
ActiveSupport::Concern
Defined in:
lib/pgbus/active_job/batch_id.rb

Overview

Gives every ActiveJob a handle on the batch it belongs to.

Two distinct ids, mirroring solid_queue's ActiveJob::BatchId:

  • batch_id — the batch this job is a member of. Assigned by the executor from the payload's pgbus_batch_id before perform, so a running job can call batch.enqueue to add siblings.
  • callback_batch_id — the batch this job reports on. Set on +on_finish+/+on_success+/+on_failure+ jobs at fire time. A callback is never a member of the batch it reports on, so its batch_id is nil.

Both round-trip through +serialize+/+deserialize+, and both are omitted from the serialized hash when unset — an unbatched job's payload is byte-for-byte what it was before this mixin existed.

Instance Method Summary collapse

Instance Method Details

#batch ⇒ Object

The batch this job reports on, or is a member of. nil when neither.



40
41
42
43
44
45
 
# File 'lib/pgbus/active_job/batch_id.rb', line 40

def batch
  return @batch if defined?(@batch)

  id = callback_batch_id || batch_id
  @batch = id && Pgbus::Batch.find(id)
end

#deserialize(job_data) ⇒ Object 



33
34
35
36
37
 
# File 'lib/pgbus/active_job/batch_id.rb', line 33

def deserialize(job_data)
  super
  self.batch_id = job_data["batch_id"]
  self.callback_batch_id = job_data["callback_batch_id"]
end

#serialize ⇒ Object 



26
27
28
29
30
31
 
# File 'lib/pgbus/active_job/batch_id.rb', line 26

def serialize
  data = super
  data["batch_id"] = batch_id if batch_id
  data["callback_batch_id"] = callback_batch_id if callback_batch_id
  data
end