Class: Karafka::Pro::ActiveJob::Dispatcher

Inherits:
ActiveJob::Dispatcher show all
Defined in:
lib/karafka/pro/active_job/dispatcher.rb

Overview

Pro dispatcher that sends the ActiveJob job to a proper topic based on the queue name and that allows to inject additional options into the producer, effectively allowing for a much better and more granular control over the dispatch and consumption process.

Instance Method Summary collapse

Instance Method Details

#dispatch(job) ⇒ Object

Parameters:

  • job (ActiveJob::Base) —

    job



49
50
51
52
53
54
55
56
57
 
# File 'lib/karafka/pro/active_job/dispatcher.rb', line 49

def dispatch(job)
  producer(job).public_send(
    fetch_option(job, :dispatch_method, DEFAULTS),
    dispatch_details(job).merge!(
      topic: job.queue_name,
      payload: ::ActiveSupport::JSON.encode(serialize_job(job))
    )
  )
end

#dispatch_at(job, timestamp) ⇒ Object

Will enqueue a job to run in the future

Parameters:

  • job (Object) —

    job we want to enqueue

  • timestamp (Time) —

    time when job should run



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/karafka/pro/active_job/dispatcher.rb', line 93

def dispatch_at(job, timestamp)
  target_message = dispatch_details(job).merge!(
    topic: job.queue_name,
    payload: ::ActiveSupport::JSON.encode(serialize_job(job))
  )

  proxy_message = Pro::ScheduledMessages.schedule(
    message: target_message,
    epoch: timestamp.to_i,
    envelope: {
      # Select the scheduled messages proxy topic
      topic: fetch_option(job, :scheduled_messages_topic, DEFAULTS)
    }
  )

  producer(job).public_send(
    fetch_option(job, :dispatch_method, DEFAULTS),
    proxy_message
  )
end

#dispatch_many(jobs) ⇒ Object

Bulk dispatches multiple jobs using the Rails 7.1+ API

Parameters:

  • jobs (Array<ActiveJob::Base>) —

    jobs we want to dispatch



61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/karafka/pro/active_job/dispatcher.rb', line 61

def dispatch_many(jobs)
  # First level is type of dispatch and second is the producer we want to use to dispatch
  dispatches = Hash.new do |hash, key|
    hash[key] = Hash.new do |hash2, key2|
      hash2[key2] = []
    end
  end

  jobs.each do |job|
    d_method = fetch_option(job, :dispatch_many_method, DEFAULTS)
    producer = producer(job)

    dispatches[d_method][producer] << dispatch_details(job).merge!(
      topic: job.queue_name,
      payload: ::ActiveSupport::JSON.encode(serialize_job(job))
    )
  end

  dispatches.each do |d_method, producers|
    producers.each do |producer, messages|
      producer.public_send(
        d_method,
        messages
      )
    end
  end
end