aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--Gemfile1
-rw-r--r--Gemfile.lock6
-rw-r--r--Rakefile4
-rw-r--r--lib/active_job/enqueuing.rb19
-rw-r--r--lib/active_job/log_subscriber.rb31
-rw-r--r--lib/active_job/queue_adapters/backburner_adapter.rb25
-rw-r--r--lib/active_job/queue_adapters/inline_adapter.rb2
-rw-r--r--lib/active_job/queue_adapters/sidekiq_adapter.rb14
-rw-r--r--lib/active_job/queue_adapters/sneakers_adapter.rb10
-rw-r--r--lib/active_job/queue_adapters/sucker_punch_adapter.rb7
-rw-r--r--test/adapters/backburner.rb3
-rw-r--r--test/cases/adapter_test.rb5
-rw-r--r--test/helper.rb2
-rw-r--r--test/support/backburner/inline.rb8
14 files changed, 96 insertions, 41 deletions
diff --git a/Gemfile b/Gemfile
index 78a5826b14..e5b9a8fec6 100644
--- a/Gemfile
+++ b/Gemfile
@@ -11,3 +11,4 @@ gem 'delayed_job'
gem 'queue_classic'
gem 'sneakers', '0.1.1.pre'
gem 'que'
+gem 'backburner'
diff --git a/Gemfile.lock b/Gemfile.lock
index 6f1ac23a94..7dda62c1fe 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -21,12 +21,17 @@ GEM
thread_safe (~> 0.1)
tzinfo (~> 1.1)
amq-protocol (1.9.2)
+ backburner (0.4.5)
+ beaneater (~> 0.3.1)
+ dante (~> 0.1.5)
+ beaneater (0.3.2)
builder (3.2.2)
bunny (1.1.9)
amq-protocol (>= 1.9.2)
celluloid (0.15.2)
timers (~> 1.1.0)
connection_pool (2.0.0)
+ dante (0.1.5)
delayed_job (4.0.1)
activesupport (>= 3.0, < 4.2)
i18n (0.6.9)
@@ -92,6 +97,7 @@ PLATFORMS
DEPENDENCIES
activejob!
+ backburner
delayed_job
que
queue_classic
diff --git a/Rakefile b/Rakefile
index 933c049a68..8ab372e08c 100644
--- a/Rakefile
+++ b/Rakefile
@@ -20,11 +20,11 @@ task :default => :test
desc 'Run all adapter tests'
task :test do
- tasks = %w(test_inline test_delayed_job test_que test_queue_classic test_resque test_sidekiq test_sneakers test_sucker_punch)
+ tasks = %w(test_inline test_delayed_job test_que test_queue_classic test_resque test_sidekiq test_sneakers test_sucker_punch test_backburner)
run_without_aborting(*tasks)
end
-%w(inline delayed_job que queue_classic resque sidekiq sneakers sucker_punch).each do |adapter|
+%w(inline delayed_job que queue_classic resque sidekiq sneakers sucker_punch backburner).each do |adapter|
Rake::TestTask.new("test_#{adapter}") do |t|
t.libs << 'test'
t.test_files = FileList['test/cases/**/*_test.rb']
diff --git a/lib/active_job/enqueuing.rb b/lib/active_job/enqueuing.rb
index 43bce125b5..fb1e004a33 100644
--- a/lib/active_job/enqueuing.rb
+++ b/lib/active_job/enqueuing.rb
@@ -2,7 +2,7 @@ require 'active_job/parameters'
module ActiveJob
module Enqueuing
- ##
+ #
# Push a job onto the queue. The arguments must be legal JSON types
# (string, int, float, nil, true, false, hash or array) or
# ActiveModel::GlobalIdentication instances. Arbitrary Ruby objects
@@ -11,11 +11,12 @@ module ActiveJob
# The return value is adapter-specific and may change in a future
# ActiveJob release.
def enqueue(*args)
- ActiveSupport::Notifications.instrument "enqueue.active_job", adapter: queue_adapter, job: self, params: args
- queue_adapter.queue self, *Parameters.serialize(args)
+ serialized_args = Parameters.serialize(args)
+ ActiveSupport::Notifications.instrument "enqueue.active_job", adapter: queue_adapter, job: self, args: serialized_args
+ queue_adapter.queue self, *serialized_args
end
- ##
+ #
# Enqueue a job to be performed at +interval+ from now.
#
# enqueue_in(1.week, "mike")
@@ -25,7 +26,7 @@ module ActiveJob
enqueue_at(interval.from_now, *args)
end
- ##
+ #
# Enqueue a job to be performed at an explicit point in time.
#
# enqueue_at(Date.tomorrow.midnight, "mike")
@@ -33,12 +34,8 @@ module ActiveJob
# Returns truthy if a job was scheduled.
def enqueue_at(timestamp, *args)
ts = timestamp.to_f
- ActiveSupport::Notifications.instrument "enqueue_at.active_job", adapter: queue_adapter, timestamp: ts, job: self, params: args
- if Time.now.to_f > ts
- queue_adapter.queue self, *Parameters.serialize(args)
- else
- queue_adapter.queue_at self, ts, *Parameters.serialize(args)
- end
+ ActiveSupport::Notifications.instrument "enqueue_at.active_job", adapter: queue_adapter, timestamp: ts, job: self, args: args
+ queue_adapter.queue_at self, ts, *Parameters.serialize(args)
end
end
end
diff --git a/lib/active_job/log_subscriber.rb b/lib/active_job/log_subscriber.rb
index 2c389ed431..eae6c30745 100644
--- a/lib/active_job/log_subscriber.rb
+++ b/lib/active_job/log_subscriber.rb
@@ -1,31 +1,22 @@
+require 'active_support/core_ext/string/filters'
+
module ActiveJob
class LogSubscriber < ActiveSupport::LogSubscriber
def enqueue(event)
- payload = event.payload
- params = payload[:params]
- adapter = payload[:adapter]
- job = payload[:job]
+ queue_name = event.payload[:adapter].name.demodulize.remove('Adapter')
+ job_name = event.payload[:job].name
+ args = event.payload[:args].any? ? ": #{event.payload[:args].inspect}" : ""
- info "ActiveJob enqueued to #{adapter.name.demodulize} job #{job.name}: #{params.inspect}"
+ info "Enqueued #{job_name} to #{queue_name}" + args
end
def enqueue_at(event)
- payload = event.payload
- params = payload[:params]
- adapter = payload[:adapter]
- job = payload[:job]
- time = payload[:timestamp]
-
- info "ActiveJob enqueued at #{time} to #{adapter.name.demodulize} job #{job.name}: #{params.inspect}"
- end
-
- def perform_error(event)
- payload = event.payload
- params = payload[:params]
- job = payload[:job]
- error = payload[:error]
+ queue_name = event.payload[:adapter].name.demodulize.remove('Adapter')
+ job_name = event.payload[:job].name
+ args = event.payload[:args].any? ? ": #{event.payload[:args].inspect}" : ""
+ time = event.payload[:timestamp]
- warn "ActiveJob caught error executing #{job} with #{params.inspect}: #{error.message}"
+ info "Enqueued #{job_name} to #{queue_name} at #{time}" + args
end
def logger
diff --git a/lib/active_job/queue_adapters/backburner_adapter.rb b/lib/active_job/queue_adapters/backburner_adapter.rb
new file mode 100644
index 0000000000..0d023d9ee7
--- /dev/null
+++ b/lib/active_job/queue_adapters/backburner_adapter.rb
@@ -0,0 +1,25 @@
+require 'backburner'
+
+module ActiveJob
+ module QueueAdapters
+ class BackburnerAdapter
+ class << self
+ def queue(job, *args)
+ Backburner::Worker.enqueue JobWrapper, [ job.name, *args ], queue: job.queue_name
+ end
+
+ def queue_at(job, timestamp, *args)
+ raise NotImplementedError
+ end
+ end
+
+ class JobWrapper
+ class << self
+ def perform(job_name, *args)
+ job_name.constantize.new.perform *Parameters.deserialize(args)
+ end
+ end
+ end
+ end
+ end
+end
diff --git a/lib/active_job/queue_adapters/inline_adapter.rb b/lib/active_job/queue_adapters/inline_adapter.rb
index 4afd4f708e..8a2c7d9d92 100644
--- a/lib/active_job/queue_adapters/inline_adapter.rb
+++ b/lib/active_job/queue_adapters/inline_adapter.rb
@@ -14,7 +14,7 @@ module ActiveJob
sleep(interval) if interval > 0
job.new.perform *Parameters.deserialize(args)
rescue => ex
- ActiveSupport::Notifications.instrument "perform_error.active_job", adapter: self, job: job, params: args, error: ex
+ ActiveJob::Base.logger "Error performing #{job}: #{ex.message}"
end
end
end
diff --git a/lib/active_job/queue_adapters/sidekiq_adapter.rb b/lib/active_job/queue_adapters/sidekiq_adapter.rb
index c6faf9d054..be6bd4ee01 100644
--- a/lib/active_job/queue_adapters/sidekiq_adapter.rb
+++ b/lib/active_job/queue_adapters/sidekiq_adapter.rb
@@ -5,12 +5,20 @@ module ActiveJob
class SidekiqAdapter
class << self
def queue(job, *args)
- JobWrapper.client_push class: JobWrapper, queue: job.queue_name, args: [ job, *args ]
+ Sidekiq::Client.push \
+ 'class' => JobWrapper,
+ 'queue' => job.queue_name,
+ 'args' => [ job, *args ],
+ 'retry' => true
end
def queue_at(job, timestamp, *args)
- job = { class: JobWrapper, queue: job.queue_name, args: [ job, *args ], at: timestamp }
- JobWrapper.client_push(job)
+ Sidekiq::Client.push \
+ 'class' => JobWrapper,
+ 'queue' => job.queue_name,
+ 'args' => [ job, *args ],
+ 'at' => timestamp,
+ 'retry' => true
end
end
diff --git a/lib/active_job/queue_adapters/sneakers_adapter.rb b/lib/active_job/queue_adapters/sneakers_adapter.rb
index ae9e49a0bf..b299b25a96 100644
--- a/lib/active_job/queue_adapters/sneakers_adapter.rb
+++ b/lib/active_job/queue_adapters/sneakers_adapter.rb
@@ -1,11 +1,17 @@
require 'sneakers'
+require 'thread'
module ActiveJob
module QueueAdapters
class SneakersAdapter
+ @mutex = Mutex.new
+
class << self
def queue(job, *args)
- JobWrapper.enqueue([job, *args])
+ @mutex.synchronize do
+ JobWrapper.from_queue job.queue_name
+ JobWrapper.enqueue [ job, *args ]
+ end
end
def queue_at(job, timestamp, *args)
@@ -16,8 +22,6 @@ module ActiveJob
class JobWrapper
include Sneakers::Worker
- self.from_queue("queue", {})
-
def work(job, *args)
job.new.perform *Parameters.deserialize(args)
end
diff --git a/lib/active_job/queue_adapters/sucker_punch_adapter.rb b/lib/active_job/queue_adapters/sucker_punch_adapter.rb
index 8125499695..79043b06e0 100644
--- a/lib/active_job/queue_adapters/sucker_punch_adapter.rb
+++ b/lib/active_job/queue_adapters/sucker_punch_adapter.rb
@@ -9,7 +9,12 @@ module ActiveJob
end
def queue_at(job, timestamp, *args)
- JobWrapper.new.async.later(secs, job, *args)
+ delay = Time.now.to_f - timestamp
+ if delay > 0
+ JobWrapper.new.async.later(delay, job, *args)
+ else
+ JobWrapper.new.async.perform(job, *args)
+ end
end
end
diff --git a/test/adapters/backburner.rb b/test/adapters/backburner.rb
new file mode 100644
index 0000000000..65d05f850b
--- /dev/null
+++ b/test/adapters/backburner.rb
@@ -0,0 +1,3 @@
+require 'support/backburner/inline'
+
+ActiveJob::Base.queue_adapter = :backburner \ No newline at end of file
diff --git a/test/cases/adapter_test.rb b/test/cases/adapter_test.rb
index 1a08fa2617..703058dacb 100644
--- a/test/cases/adapter_test.rb
+++ b/test/cases/adapter_test.rb
@@ -43,4 +43,9 @@ class AdapterTest < ActiveSupport::TestCase
ActiveJob::Base.queue_adapter = :sneakers
assert_equal ActiveJob::QueueAdapters::SneakersAdapter, ActiveJob::Base.queue_adapter
end
+
+ test 'should load Backburner adapter' do
+ ActiveJob::Base.queue_adapter = :backburner
+ assert_equal ActiveJob::QueueAdapters::BackburnerAdapter, ActiveJob::Base.queue_adapter
+ end
end
diff --git a/test/helper.rb b/test/helper.rb
index fc3e2642df..a5adc882d9 100644
--- a/test/helper.rb
+++ b/test/helper.rb
@@ -7,3 +7,5 @@ require 'active_job'
require "adapters/#{ENV['AJADAPTER'] || 'inline'}"
require 'active_support/testing/autorun'
+
+ActiveJob::Logging.logger.level = Logger::ERROR
diff --git a/test/support/backburner/inline.rb b/test/support/backburner/inline.rb
new file mode 100644
index 0000000000..f761b53e27
--- /dev/null
+++ b/test/support/backburner/inline.rb
@@ -0,0 +1,8 @@
+require 'backburner'
+
+Backburner::Worker.class_eval do
+ class << self; alias_method :original_enqueue, :enqueue; end
+ def self.enqueue(job_class, args=[], opts={})
+ job_class.perform(*args)
+ end
+end \ No newline at end of file
generated by cgit v1.2.3 (git 2.25.1) at 2024-07-16 08:47:00 +0000