Module: Pgbus::Concurrency

Extended by:
ActiveSupport::Concern
Defined in:
lib/pgbus/concurrency.rb,
lib/pgbus/concurrency/semaphore.rb,
lib/pgbus/concurrency/blocked_execution.rb

Defined Under Namespace

Modules: BlockedExecution, Semaphore

Constant Summary collapse

METADATA_KEY = 
"pgbus_concurrency_key"

Class Method Summary collapse

Class Method Details

.extract_key(payload) ⇒ Object

Extract the concurrency key from a deserialized payload.



71
72
73
 
# File 'lib/pgbus/concurrency.rb', line 71

def extract_key(payload)
  payload[METADATA_KEY]
end

.inject_metadata(active_job, payload_hash) ⇒ Object

Inject the resolved concurrency key into the job's serialized payload.



63
64
65
66
67
68
 
# File 'lib/pgbus/concurrency.rb', line 63

def inject_metadata(active_job, payload_hash)
  key = resolve_key(active_job)
  return payload_hash unless key

  payload_hash.merge(METADATA_KEY => key)
end

.resolve_key(active_job) ⇒ Object

Resolve the concurrency key for a given job instance. Returns nil if the job class has no concurrency config.



49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/pgbus/concurrency.rb', line 49

def resolve_key(active_job)
  return nil unless active_job.class.respond_to?(:pgbus_concurrency)

  config = active_job.class.pgbus_concurrency
  return nil unless config

  # Class-name default, resolved from the ENQUEUED job's class so an
  # inherited declaration keys each subclass separately (#357).
  return active_job.class.name unless config[:key]

  Support.call_key_proc(config[:key], active_job.arguments)
end