merge master

7 files changed, 69 insertions, 39 deletions

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