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



41
42
43
44
45
46
47
48
49
 
# File 'lib/karafka/pro/active_job/dispatcher.rb', line 41

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



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

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



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

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