Module: Zizq::ActiveJobConfig

Includes:
JobConfig
Defined in:
lib/zizq/active_job_config.rb,
sig/generated/zizq/active_job_config.rbs

Overview

Zizq configuration DSL for ActiveJob classes.

Extend this module in an ActiveJob subclass to allow enqueueing jobs via Zizq.enqueue and to gain access to Zizq features like unique jobs, backoff, and retention:

class SendEmailJob < ApplicationJob
extend Zizq::ActiveJobConfig

zizq_unique true, scope: :active
zizq_backoff exponent: 4.0, base: 15, jitter: 30

def perform(user_id, template:)
  # ...
end
end

Serialization uses ActiveJob's own format so that GlobalID, Time, and other ActiveJob-supported types are handled correctly. The Zizq worker must use the ActiveJob dispatcher:

Zizq.configure do |c|
c.dispatcher = ActiveJob::QueueAdapters::ZizqAdapter::Dispatcher
end

Instance Method Summary collapse

Methods included from JobConfig

#batch_for_enqueue, #build_batch_expressions, #hash_args, #name, #normalize_payload, #zizq_backoff, #zizq_batch_arg, #zizq_batch_dedup?, #zizq_batch_key, #zizq_batch_kwarg, #zizq_batch_limit, #zizq_batch_sorted?, #zizq_batched, #zizq_enqueue_request, #zizq_priority, #zizq_retention, #zizq_retry_limit, #zizq_unique, #zizq_unique_key, #zizq_unique_scope

Instance Method Details

#hashable_args(*args, **kwargs) ⇒ Object

Hash only the arguments portion of the serialized ActiveJob payload. The full envelope contains volatile fields (job_id, enqueued_at, etc.) that change per instance and would defeat both zizq_unique_key and zizq_batch_key.

Parameters:

  • args (Object)
  • kwargs (Object)

Returns:

  • (Object) 


159
160
161
 
# File 'lib/zizq/active_job_config.rb', line 159

def hashable_args(*args, **kwargs) #: (*untyped, **untyped) -> untyped
  new(*args, **kwargs).serialize["arguments"]
end

#new ⇒ Object

ActiveJob::Base.new — invisible to steep without this.

Parameters:

  • (Object)
  • (Object)

Returns:

  • (Object) 


32
 
# File 'sig/generated/zizq/active_job_config.rbs', line 32

def new: (*untyped, **untyped) -> untyped

#queue_name ⇒ String?

ActiveJob::Base.queue_name — invisible to steep without this.

Returns:

  • (String, nil) 


35
 
# File 'sig/generated/zizq/active_job_config.rbs', line 35

def queue_name: () -> String?

#zizq_batch_expressions ⇒ Object

Static jq expressions for the batched-job configuration on this class. Targets .arguments[N] for a positional batch arg or .arguments[-1].NAME for a keyword batch arg — the latter assumes the class actually receives kwargs so that the last argument is the ActiveJob kwargs hash. Behaviour is unspecified if the declared kwarg: isn't present at enqueue time.

Returns:

  • (Object) 


140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/zizq/active_job_config.rb', line 140

def zizq_batch_expressions #: () -> Zizq::batch_expressions?
  return nil unless zizq_batched

  target =
    if (idx = zizq_batch_arg)
      ".arguments[#{idx}]"
    else
      ".arguments[-1].#{zizq_batch_kwarg}"
    end

  build_batch_expressions(target)
end

#zizq_deserialize(_payload) ⇒ Object

Deserialization is handled by ActiveJob::Base.execute on the worker side. This method is not used in the ActiveJob dispatch path.

Parameters:

  • _payload (Object)

Returns:

  • (Object) 


66
67
68
69
 
# File 'lib/zizq/active_job_config.rb', line 66

def zizq_deserialize(_payload) #: (untyped) -> [Array[untyped], Hash[Symbol, untyped]]
  raise NotImplementedError,
        "ActiveJob handles deserialization via ActiveJob::Base.execute"
end

#zizq_payload_filter(*args, **kwargs) ⇒ Object

Generate a jq expression that exactly matches payloads with the given arguments.

This is used for filtering in Zizq::Query.

Generates an expression of the form:

.arguments == ["a","b",{"example":true,"_aj_ruby2_keywords":["example"]}]

Parameters:

  • args (Object)
  • kwargs (Object)

Returns:

  • (Object) 


79
80
81
82
 
# File 'lib/zizq/active_job_config.rb', line 79

def zizq_payload_filter(*args, **kwargs) #: (*untyped, **untyped) -> String
  arguments = zizq_serialize(*args, **kwargs)["arguments"]
  ".arguments == #{JSON.generate(arguments)}"
end

#zizq_payload_subset_filter(*args, **kwargs) ⇒ Object

Generate a jq expression that matches jobs whose positional args start with the given values and whose kwargs contain the given key/value pairs.

This is used for filtering in Zizq::Query.

Generates expressions of the form:

(.arguments[0:2] == ["a","b"])

or

(.arguments[0:2] == ["a","b"]) and
(.arguments[-1] | has("_aj_ruby2_keywords")) and
(.arguments[-1] | contains({"example":true}))

Parameters:

  • args (Object)
  • kwargs (Object)

Returns:

  • (Object) 


99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/zizq/active_job_config.rb', line 99

def zizq_payload_subset_filter(*args, **kwargs) #: (*untyped, **untyped) -> String
  arguments = zizq_serialize(*args, **kwargs)["arguments"]

  # ActiveJob flattens arguments into a single array, but marks kwargs with
  # "_aj_ruby2_keywords" => ["key1", "key2", ...] in the last element of
  # the array where kwargs are present. We need to detect this to generate
  # a suitable expression.
  serialized_args, serialized_kwargs =
    if arguments.size > 0
      # See what the last argument looks like. It might be kwargs.
      maybe_kwargs = arguments.pop

      # If it's got "_aj_ruby2_keywords" then it is kwargs.
      if maybe_kwargs.is_a?(Hash) && maybe_kwargs["_aj_ruby2_keywords"]
        # We only want the actual kwargs, not the marker.
        [arguments, maybe_kwargs.except("_aj_ruby2_keywords")]
      else
        # It wasn't kwargs, so put it back.
        [arguments.push(maybe_kwargs), nil]
      end
    else
      [arguments, nil]
    end

  parts = [] #: Array[String]
  parts << %Q<(.arguments[0:#{serialized_args.size}] == #{JSON.generate(serialized_args)})>

  if serialized_kwargs
    parts << %Q<(.arguments[-1] | has("_aj_ruby2_keywords"))>
    parts << %Q<(.arguments[-1] | contains(#{JSON.generate(serialized_kwargs)}))>
  end

  parts.join(" and ")
end

#zizq_queue(name = nil) ⇒ Object

Use ActiveJob's queue_name as the default queue, falling back to any explicit zizq_queue setting, then "default".

Parameters:

  • name (Object) (defaults to: nil)

Returns:

  • (Object) 


46
47
48
49
50
51
52
 
# File 'lib/zizq/active_job_config.rb', line 46

def zizq_queue(name = nil) #: (?String?) -> String
  if name
    super
  else
    @zizq_queue || queue_name || "default"
  end
end

#zizq_serialize(*args, **kwargs) ⇒ Object

Serialize using ActiveJob's own format.

Creates a temporary ActiveJob instance to produce the canonical serialized form. Returns the full serialized hash (including job_class, arguments, queue_name, etc.) so that the payload stored in Zizq matches what ActiveJob::Base.execute expects.

Parameters:

  • args (Object)
  • kwargs (Object)

Returns:

  • (Object) 


60
61
62
 
# File 'lib/zizq/active_job_config.rb', line 60

def zizq_serialize(*args, **kwargs) #: (*untyped, **untyped) -> Hash[String, untyped]
  new(*args, **kwargs).serialize
end