Module: Axn::Async::BatchEnqueue::DSL

Defined in:
lib/axn/async/batch_enqueue.rb

Overview

DSL methods for batch enqueueing

Instance Method Summary collapse

Instance Method Details

#enqueue_all(**static_args) ⇒ String

Batch enqueue jobs for this action.

Validates async is configured, validates static args, then executes iteration asynchronously via EnqueueAllOrchestrator.

Fields with model: declarations are automatically inferred for iteration. You can override iteration by passing enumerables (to replace source) or scalars (to make fields static) as kwargs.

Parameters:

  • static_args (Hash) —

    Arguments to pass to every enqueued job.

    • Scalar values: Treated as static args (passed to all jobs)
    • Enumerable values: Treated as iteration sources (overrides configured sources)
    • Exception: Arrays/Sets are static when field expects enumerable type

Returns:

  • (String) —

    Job ID from the async adapter

Raises:

  • (NotImplementedError) —

    If async is not configured

  • (MissingEnqueuesEachError) —

    If expects exist but no iteration config found

  • (ArgumentError) —

    If required static fields are missing



76
77
78
 
# File 'lib/axn/async/batch_enqueue.rb', line 76

def enqueue_all(**static_args)
  EnqueueAllOrchestrator.enqueue_for(self, **static_args)
end

#enqueues_each(field, from: nil, via: nil, &filter_block) ⇒ Object

Declare a field to iterate over for batch enqueueing.

Note: Fields with model: declarations are automatically inferred, so enqueues_each is only needed to override defaults, add filtering, or iterate non-model fields.

Parameters:

  • field (Symbol) —

    The field name from expects to iterate over

  • from (Proc, Symbol, nil) (defaults to: nil) —

    The source collection.

    • Proc/lambda: Called to get the collection
    • Symbol: Method name on the action class
    • nil: Inferred from field's model: declaration (Model.all)
  • via (Symbol, nil) (defaults to: nil) —

    Optional attribute to extract from each item (e.g., :id)

  • block (Proc, nil) —

    Optional filter block - return truthy to enqueue, falsy to skip



93
94
95
 
# File 'lib/axn/async/batch_enqueue.rb', line 93

def enqueues_each(field, from: nil, via: nil, &filter_block)
  self._batch_enqueue_configs += [Config.new(field:, from:, via:, filter_block:)]
end

#on_enqueue_all(handler = nil) {|count, sources| ... } ⇒ Object

Register a once-per-run callback that fires after the batch fan-out completes.

Runs inside EnqueueAllOrchestrator (off the clock thread), after all jobs are enqueued. The handler is evaluated in the context of this action class, so it has access to class-level log/info/warn (a Symbol handler resolves to a class method). Like the other on_* callbacks, this accepts a block or a Symbol method name, supports if:/unless:, and — when multiple are declared — fires them most-recent-first (last-defined wins).

The handler may declare any subset of these keyword arguments (or none): Note: if:/unless: conditions are evaluated like the other callbacks' matchers (against the action with no exception); they cannot observe sources/count.

A raise inside the handler is swallowed (logged; re-raised in dev only when Axn.config.best_effort_raises_in_dev is set) and cannot change the enqueue outcome — rescue inside your handler if you need stronger guarantees.

Examples:

Post a run summary to Slack

on_enqueue_all do |sources:, count:|
  active, inactive = sources[:tax_profile].partition { _1.user.active? }
  SlackSender.call(channel: :eng_ops, text: "#{active.size} active, #{inactive.size} deactivated (#{count} enqueued)")
end

Count-only heartbeat via a class method

on_enqueue_all :log_summary
def self.log_summary(count:) = info "Found #{count} events"

Yield Parameters:

  • count (Integer) —

    exact number of jobs enqueued (post-filter)

  • sources (Hash{Symbol => Object}) —

    resolved (un-materialized) source per iterated field, e.g. { tax_profile: } or { user: , company: }



127
 
# File 'lib/axn/async/batch_enqueue.rb', line 127

def on_enqueue_all(handler = nil, **, &block) = _add_callback(:enqueue_all, handler:, **, block:)