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



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

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



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

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



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
108
 
# File 'lib/karafka/pro/active_job/dispatcher.rb', line 82

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