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



45
46
47
48
49
50
51
52
53
 
# File 'lib/karafka/pro/active_job/dispatcher.rb', line 45

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



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

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



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

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