Class: ActiveJob::QueueAdapters::HutchAdapter

Inherits:
Object
  • Object
show all
Defined in:
lib/active_job/queue_adapters/hutch_adapter.rb

Overview

Hutch adapter for Active Job

Read more about Hutch here.

Rails.application.config.active_job.queue_adapter = :hutch

Constant Summary collapse

AJ_ROUTING_KEY =

All activejob Message will routing to one RabbitMQ Queue. Because Hutch will one Consumer per Queue

"active_job"

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ HutchAdapter

Returns a new instance of HutchAdapter.



15
16
17
 
# File 'lib/active_job/queue_adapters/hutch_adapter.rb', line 15

def initialize
  @monitor = Monitor.new
end

Class Method Details

.register_actice_job_classes ⇒ Object

Register all ActiveJob Class to Hutch. (per queue per consumer)



45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/active_job/queue_adapters/hutch_adapter.rb', line 45

def self.register_actice_job_classes
  # TODO: 需要考虑如何将 AJ 的 Proc queue_name 动态注册到 Hutch
  queue_consumers = {}

  Dir.glob(Rails.root.join('app/jobs/**/*.rb')).each { |x| require_dependency x }
  ActiveJob::Base.descendants.each do |job_clazz|
    # Need activeJob instance #queue_name
    job = job_clazz.new
    # Multi queue only have one consumer
    next if queue_consumers.key?(job.queue_name)
    queue_consumers[job.queue_name] = HutchAdapter.dynamic_consumer(job)
    Hutch.register_consumer(queue_consumers[job.queue_name])
  end
end

.routing_key(job) ⇒ Object

Get an routing_key



40
41
42
 
# File 'lib/active_job/queue_adapters/hutch_adapter.rb', line 40

def self.routing_key(job)
  "#{AJ_ROUTING_KEY}.#{job.queue_name}"
end

Instance Method Details

#enqueue(job) ⇒ Object

:nodoc:



19
20
21
22
23
24
 
# File 'lib/active_job/queue_adapters/hutch_adapter.rb', line 19

def enqueue(job) #:nodoc:
  @monitor.synchronize do
    # publish all job data to hutch
    Hutch.publish(HutchAdapter.routing_key(job), job.serialize)
  end
end

#enqueue_at(job, timestamp) ⇒ Object

:nodoc:



26
27
28
29
 
# File 'lib/active_job/queue_adapters/hutch_adapter.rb', line 26

def enqueue_at(job, timestamp) #:nodoc:
  interval = [(timestamp - Time.now.utc.to_i), 1.second].max
  enqueue_in(interval, job.serialize, HutchAdapter.routing_key(job))
end

#enqueue_in(interval, message, routing_key) ⇒ Object 



31
32
33
34
35
36
37
 
# File 'lib/active_job/queue_adapters/hutch_adapter.rb', line 31

def enqueue_in(interval, message, routing_key)
  @monitor.synchronize do
    # must be integer
    props = { expiration: interval.in_milliseconds.to_i }
    Hutch::Schedule.publish(routing_key, message, props)
  end
end