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



69
70
71
72
73
74
75
76
77
 
# File 'lib/karafka/pro/active_job/dispatcher.rb', line 69

def dispatch(job)
  producer(job).public_send(
    fetch_option(job, :dispatch_method, DEFAULTS),
    dispatch_details(job).merge!(
      topic: job.queue_name,
      payload: 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



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/karafka/pro/active_job/dispatcher.rb', line 113

def dispatch_at(job, timestamp)
  # If request is in the past, we can dispatch it directly without any envelope wrapping
  # and proxy pass via scheduled messages. This improves latency and also allows for things
  # like immediate continuation to work without having to have scheduled messages
  return dispatch(job) if timestamp.to_f <= Time.now.to_f

  target_message = dispatch_details(job).merge!(
    topic: job.queue_name,
    payload: 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



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/karafka/pro/active_job/dispatcher.rb', line 81

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: serialize_job(job)
    )
  end

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