merge master

5 files changed, 50 insertions, 8 deletions

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