Module: Zizq

Defined in:
lib/zizq/test.rb,
lib/zizq.rb,
lib/zizq/job.rb,
lib/zizq/error.rb,
lib/zizq/query.rb,
lib/zizq/client.rb,
lib/zizq/router.rb,
lib/zizq/worker.rb,
lib/zizq/backoff.rb,
lib/zizq/crontab.rb,
lib/zizq/version.rb,
lib/zizq/lifecycle.rb,
lib/zizq/resources.rb,
lib/zizq/job_config.rb,
lib/zizq/middleware.rb,
lib/zizq/test/client.rb,
lib/zizq/bulk_enqueue.rb,
lib/zizq/enqueue_with.rb,
lib/zizq/ack_processor.rb,
lib/zizq/configuration.rb,
lib/zizq/crontab_entry.rb,
lib/zizq/resources/job.rb,
lib/zizq/resources/page.rb,
lib/zizq/crontab_builder.rb,
lib/zizq/enqueue_request.rb,
lib/zizq/active_job_config.rb,
lib/zizq/tls_configuration.rb,
lib/zizq/resources/job_page.rb,
lib/zizq/resources/resource.rb,
lib/zizq/resources/cron_entry.rb,
lib/zizq/resources/cron_group.rb,
lib/zizq/resources/error_page.rb,
lib/zizq/worker_configuration.rb,
lib/zizq/crontab_entry_builder.rb,
lib/zizq/resources/error_record.rb,
lib/zizq/resources/job_template.rb,
lib/zizq/resources/error_enumerator.rb,
sig/zizq.rbs,
sig/generated/zizq.rbs,
sig/generated/zizq/job.rbs,
sig/generated/zizq/test.rbs,
sig/generated/zizq/error.rbs,
sig/generated/zizq/query.rbs,
sig/generated/zizq/client.rbs,
sig/generated/zizq/router.rbs,
sig/generated/zizq/worker.rbs,
sig/generated/zizq/backoff.rbs,
sig/generated/zizq/crontab.rbs,
sig/generated/zizq/version.rbs,
sig/generated/zizq/lifecycle.rbs,
sig/generated/zizq/job_config.rbs,
sig/generated/zizq/middleware.rbs,
sig/generated/zizq/test/client.rbs,
sig/generated/zizq/bulk_enqueue.rbs,
sig/generated/zizq/enqueue_with.rbs,
sig/generated/zizq/ack_processor.rbs,
sig/generated/zizq/configuration.rbs,
sig/generated/zizq/crontab_entry.rbs,
sig/generated/zizq/resources/job.rbs,
sig/generated/zizq/resources/page.rbs,
sig/generated/zizq/crontab_builder.rbs,
sig/generated/zizq/enqueue_request.rbs,
sig/generated/zizq/active_job_config.rbs,
sig/generated/zizq/tls_configuration.rbs,
sig/generated/zizq/resources/job_page.rbs,
sig/generated/zizq/resources/resource.rbs,
sig/generated/zizq/resources/cron_entry.rbs,
sig/generated/zizq/resources/cron_group.rbs,
sig/generated/zizq/resources/error_page.rbs,
sig/generated/zizq/worker_configuration.rbs,
sig/generated/zizq/crontab_entry_builder.rbs,
sig/generated/zizq/resources/error_record.rbs,
sig/generated/zizq/resources/job_template.rbs,
sig/generated/zizq/resources/error_enumerator.rbs

Overview

rbs_inline: enabled frozen_string_literal: true

Defined Under Namespace

Modules: ActiveJobConfig, Job, JobConfig, Middleware, RESET, Resources, Test, UNCHANGED, _Dispatcher, _EnqueueTarget, _JobClass, _ToF Classes: AckProcessor, Backoff, BulkEnqueue, Client, ClientError, Configuration, ConnectionError, Crontab, CrontabBuilder, CrontabEntry, CrontabEntryBuilder, EnqueueRequest, EnqueueWith, Error, Lifecycle, NotFoundError, Query, ResponseError, Router, ServerError, StreamError, TlsConfiguration, Worker, WorkerConfiguration

Constant Summary collapse

VERSION =

Signature:

  • String

Returns:

  • (String) 
"0.6.0"

Class Method Summary collapse

Class Method Details

.build_enqueue_request(job_class, *args, **kwargs, &block) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Build an EnqueueRequest for a single job class enqueue.

RBS:

  • job_class: Class & Zizq::JobConfig

  • args: Array[untyped]

  • kwargs: Hash[Symbol, untyped]

  • &block: ?(EnqueueRequest) -> void

  • return: EnqueueRequest

Parameters:



363
364
365
366
367
368
369
370
371
372
373
 
# File 'lib/zizq.rb', line 363

def build_enqueue_request(job_class, *args, **kwargs, &block)
  unless job_class.is_a?(Class) && job_class.is_a?(Zizq::JobConfig)
    raise ArgumentError,
          "#{job_class.inspect} must include Zizq::Job or extend Zizq::ActiveJobConfig"
  end

  zizq_job_class = job_class #: Zizq::JobConfig
  req = zizq_job_class.zizq_enqueue_request(*args, **kwargs)
  yield req if block_given?
  req
end

.client ⇒ Object

Returns a shared client instance built from the global configuration.

The client is memoized so that persistent HTTP connections are reused across calls, reducing TCP connection overhead.

When configuration.test_mode is set, a Zizq::Test::Client is returned instead — buffering enqueues in memory rather than talking to a real server.

Returns:

  • (Object) 


85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/zizq.rb', line 85

def client #: () -> Client
  @client ||=
    begin
      @client_mutex.synchronize do
        break @client if @client

        configuration.validate!
        @client =
          if configuration.test_mode
            Test::Client.new
          else
            Client.new(
              url: configuration.url,
              format: configuration.format,
              ssl_context: configuration.ssl_context,
              read_timeout: configuration.read_timeout,
              stream_idle_timeout: configuration.stream_idle_timeout
            )
          end
      end
    end
end

.configuration ⇒ Object

Returns the client configuration.

The configuration can be updated by calling [Zizq::configure].

This configuration is for the client only. Worker parameters are configured on a per-run basis for flexibility.

Returns:

  • (Object) 


56
57
58
 
# File 'lib/zizq.rb', line 56

def configuration #: () -> Configuration
  @configuration ||= Configuration.new
end

.configure ⇒ Object

Yields the global configuration ready for updates, which should be done during application initialization, before any jobs are enqueued or worked.

Zizq.configure do |c|
c.url = "http://localhost:7890"
c.format = :msgpack
c.dequeue_middleware.use(MyDequeueMiddleware.new)
end

Returns:

  • (Object) 


69
70
71
72
73
74
75
 
# File 'lib/zizq.rb', line 69

def configure #: () { (Configuration) -> void } -> void
  yield configuration
ensure
  # shared client is potentially stale
  @client&.close
  @client = nil
end

.crontab(name) ⇒ Zizq::Crontab

Obtain a handle for the given Crontab schedule.

This is a lazy operation. The schedule data is only fetched from the Zizq server upon first accessing data within the schedule.

Zizq.crontab("default").paused?
Zizq.crontab("default").resume!

RBS:

  • name: String

  • return: Zizq::Crontab

Parameters:

  • name (String)

Returns:



219
220
221
 
# File 'lib/zizq.rb', line 219

def crontab(name)
  Crontab.new(name)
end

.crontabs ⇒ Object

Return a list of all available Crontab schedules.

Returns:

  • (Object) 


142
143
144
 
# File 'lib/zizq.rb', line 142

def crontabs #: () -> Array[String]
  Zizq.client.list_cron_groups
end

.define_crontab(name, timezone: nil, paused: nil) {|arg0| ... } ⇒ Zizq::Crontab

Define (or redefine) a Crontab schedule.

This requires a Pro license on the Zizq server.

Crontabs are used to define collections of recurring jobs that run on a specified schedule, such as at 2am on every Monday. Each entry on the Crontab is a single job enqueue, which the Zizq server automatically triggers at the correct point in time. Zizq uses standard Cron expression syntax (with support for seconds via 6-fields) to define entries.

This is designed to be idempotent. You can define a schedule somewhere in your application startup process (after Zizq.configure) and it doesn't matter if multiple process all define the same schedule. Zizq is smart enough to handle this correctly.

Entire schedules, and individual entries on a schedule, can be paused and resumed.

By default schedules operate in the system time zone of the Zizq server but an explicit IANA timezone name can be specified when defining the Crontab.

This method sends exactly one request to the Zizq server upon completion of the block. Any existing entries are retained. Any new entries are added, any absent entries are removed, and any modified entries are replaced. In short, whatever the block defines is what the entire resulting Crontab schedule will look like.

Zizq.define_crontab("example", timezone: "Europe/Rome") do |cron|
cron.define_entry(
  "refresh_data_warehose",
  "*/15 * * * *"
).enqueue(RefreshDataWarehoseJob, incremental: true)

cron.define_entry(
  "expire_acess_tokens",
  "*/10 * * * * *"
).enqueue_raw(
  queue: "identity-server",
  type: "expire_access_tokens",
  priority: 100,
  payload: {},
)
end

When jobs are pushed to the queue at their execution time, Zizq handles this atomically, so there is no risk of a duplicate enqueue for the same schedule tick. However, if you have long-running jobs that should not be permitted to overlap, such as in the case your schedule runs every 10 seconds but jobs can take 30 seconds to execute, you should consider using unique jobs.

RBS:

  • name: String

  • timezone: String?

  • paused: bool?

  • &block: (Zizq::CrontabBuilder) -> void

  • return: Zizq::Crontab

Parameters:

  • name (String)
  • timezone: (String, nil) (defaults to: nil)
  • paused: (Boolean, nil) (defaults to: nil)

Yields:

Yield Parameters:

Yield Returns:

  • (void)

Returns:



203
204
205
206
207
 
# File 'lib/zizq.rb', line 203

def define_crontab(name, timezone: nil, paused: nil, &block)
  crontab = Crontab.new(name)
  crontab.redefine(timezone:, paused:, &block)
  crontab
end

.enqueue(job_class, *args, **kwargs, &block) ⇒ void

This method returns an undefined value.

Enqueue a job by class with positional and keyword arguments.

By default all arguments are serialized as JSON, which means hashes with symbol keys will become hashes with string keys. The serialization behaviour can be changed by implementing ::zizq_serialize and ::zizq_deserialize as class methods on the job class.

Default job options can be overridden at enqueue-time by providing a block which receives a mutable Zizq::EnqueueRequest instance.

Zizq.enqueue(SendEmailJob, 42, template: "welcome")
Zizq.enqueue(SendEmailJob, 42) { |o| o.queue = "priority" }

Job classes may also override ::zizq_enqueue_options to implement dynamically computed options, such as dynamic prioritisation. This class method accepts the same arguments as the #perform method and returns an instance of Zizq::EnqueueRequest. Any overrides may call super and modify the result.

class SendEmailJob
include Zizq::Job

zizq_priority 1000

def self.zizq_enqueue_options(user_id, template:)
  opts = super
  opts.priority /= 2 if template == "welcome"
  opts
end

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

RBS:

  • job_class: Class & Zizq::JobConfig

  • args: Array[untyped]

  • kwargs: Hash[Symbol, untyped]

  • &block: ?(EnqueueRequest) -> void

  • return: Resources::Job

Parameters:



263
264
265
266
267
 
# File 'lib/zizq.rb', line 263

def enqueue(job_class, *args, **kwargs, &block)
  req = build_enqueue_request(job_class, *args, **kwargs, &block)
  req = configuration.enqueue_middleware.call(req)
  client.enqueue(**req.to_enqueue_params)
end

.enqueue_bulk {|arg0| ... } ⇒ Array[Resources::Job]

RBS:

  • &block: (BulkEnqueue) -> void

  • return: Array[Resources::Job]

Yields:

Yield Parameters:

Yield Returns:

  • (void)

Returns:



342
343
344
345
346
347
348
349
350
351
352
353
 
# File 'lib/zizq.rb', line 342

def enqueue_bulk(&block)
  builder = BulkEnqueue.new
  yield builder
  return [] if builder.requests.empty?

  jobs =
    builder.requests.map do |req|
      configuration.enqueue_middleware.call(req).to_enqueue_params
    end

  client.enqueue_bulk(jobs:)
end

.enqueue_raw(queue:, type:, payload:, **opts) ⇒ Resources::Job

Enqueue a job by providing raw inputs to the Zizq server.

This is for advanced use cases such as enqueueing jobs for consumption in other programming languages.

Zizq.enqueue_raw(
queue: "emails",
type: "send_email",
payload: {user_id: 42, template: "welcome"}
)

If using this method to enqueue a job that is intended for consumption in the Ruby client itself a custom dispatcher implementation is likely required:

Zizq.configure do |c|
c.dispatcher = MyDispatcher.new
end

RBS:

  • queue: String

  • type: String

  • payload: untyped

  • priority: Integer?

  • ready_at: Zizq::to_f?

  • retry_limit: Integer?

  • backoff: Zizq::backoff?

  • retention: Zizq::retention?

  • unique_key: String?

  • unique_while: Zizq::unique_scope?

  • batch: Zizq::batch?

  • return: Resources::Job

Parameters:

  • queue: (String)
  • type: (String)
  • payload: (Object)
  • opts (Object)

Returns:



300
301
302
303
304
 
# File 'lib/zizq.rb', line 300

def enqueue_raw(queue:, type:, payload:, **opts)
  req = EnqueueRequest.new(queue:, type:, payload:, **opts)
  req = configuration.enqueue_middleware.call(req)
  client.enqueue(**req.to_enqueue_params)
end

.enqueue_with(**overrides) ⇒ EnqueueWith

Enqueue multiple jobs atomically in a single bulk request.

This can significantly imprive throughput when many jobs need to be enqueued collectively. There is no upper limit on the number of jobs in the request though generally it is probably wise to keep this to less than 1000 jobs unless you have strong atomicity requuirements for a larger number of jobs..

Yields a builder object whose #enqueue method accepts the same arguments as Zizq.enqueue. All collected jobs are sent as a single POST /jobs/bulk request and an array of jobs is returned in the same order as the inputs.

Zizq.enqueue_bulk do |b|
b.enqueue(ProcessPaymentJob, 7)
b.enqueue(SendEmailJob, 42, template: "welcome")
b.enqueue(SendEmailJob, 42) { |o| o.queue = "priority" }
end

Build a scoped enqueue helper that applies the given option overrides to every enqueue routed through it. Equivalent to using the block form of Zizq.enqueue, but composable and reusable.

Zizq.enqueue_with(ready_at: Time.now + 3600).enqueue(MyJob, 42)
Zizq.enqueue_with(priority: 0).enqueue_bulk { |b| ... }

See Zizq::EnqueueWith for details.

RBS:

  • overrides: Hash[Symbol, untyped]

  • return: EnqueueWith

Parameters:

  • overrides (Object)

Returns:



336
337
338
 
# File 'lib/zizq.rb', line 336

def enqueue_with(**overrides)
  EnqueueWith.new(self, overrides)
end

.query ⇒ Zizq::Query

Start a query to retrieve or modify job data.

RBS:

  • id: (String | Array[String])?

  • queue: (String | Array[String])?

  • type: (String | Array[String])?

  • status: (String | Array[String])?

  • jq_filter: String?

  • order: Zizq::sort_direction?

  • limit: Integer?

  • page_size: Integer?

  • return: Zizq::Query

Returns:



137
138
139
 
# File 'lib/zizq.rb', line 137

def query(...)
  Query.new(...)
end

.queues ⇒ Object

List all distinct queue names on the server.

Returns:

  • (Object) 


122
123
124
 
# File 'lib/zizq.rb', line 122

def queues #: () -> Array[String]
  client.get_queues
end

.reset! ⇒ Object

Resets all global state: configuration and shared client. Intended for use in tests.

Returns:

  • (Object) 


110
111
112
113
114
 
# File 'lib/zizq.rb', line 110

def reset! #: () -> void
  @client&.close
  @client = nil
  @configuration = nil
end

.server_version ⇒ Object

Server version string.

Returns:

  • (Object) 


117
118
119
 
# File 'lib/zizq.rb', line 117

def server_version #: () -> String
  client.server_version
end