merge master

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