Don't deserialize parameters in individual adapters.

It's not their responsibility and this makes logging (and rescuing from errors, eventually!) a lot easier.

13 files changed, 34 insertions, 14 deletions

diff --git a/lib/active_job/base.rb b/lib/active_job/base.rb
index 3d16f38275..8bddfde09f 100644
--- a/lib/active_job/base.rb
+++ b/lib/active_job/base.rb
@@ -1,6 +1,7 @@
 require 'active_job/queue_adapter'
 require 'active_job/queue_name'
 require 'active_job/enqueuing'
+require 'active_job/performing'
 require 'active_job/logging'
 
 module ActiveJob
@@ -8,6 +9,7 @@ module ActiveJob
     extend QueueAdapter
     extend QueueName
     extend Enqueuing
+    include Performing
     extend Logging
   end
 end
diff --git a/lib/active_job/enqueuing.rb b/lib/active_job/enqueuing.rb
index 7f2aef871e..dd4556c237 100644
--- a/lib/active_job/enqueuing.rb
+++ b/lib/active_job/enqueuing.rb
@@ -32,7 +32,7 @@ module ActiveJob
     def enqueue_at(timestamp, *args)
       timestamp       = timestamp.to_f
       serialized_args = Parameters.serialize(args)
-      ActiveSupport::Notifications.instrument "enqueue_at.active_job", adapter: queue_adapter, timestamp: timestamp, job: self, args: serialized_args
+      ActiveSupport::Notifications.instrument "enqueue_at.active_job", adapter: queue_adapter, job: self, args: serialized_args, timestamp: timestamp
       queue_adapter.queue_at self, timestamp, *serialized_args
     end
   end
diff --git a/lib/active_job/logging.rb b/lib/active_job/logging.rb
index f06f12087b..22748e3d5f 100644
--- a/lib/active_job/logging.rb
+++ b/lib/active_job/logging.rb
@@ -10,7 +10,11 @@ module ActiveJob
       end
 
       def enqueue_at(event)
-        info "Enqueued #{event.payload[:job].name} to #{queue_name(event)} at #{enqueud_at(event)}" + args_info(event)
+        info "Enqueued #{event.payload[:job].name} to #{queue_name(event)} at #{enqueued_at(event)}" + args_info(event)
+      end
+      
+      def perform(event)
+        info "Performed #{event.payload[:job].name} to #{queue_name(event)}" + args_info(event)
       end
 
       private
@@ -22,7 +26,7 @@ module ActiveJob
           event.payload[:args].any? ? ": #{event.payload[:args].inspect}" : ""
         end
         
-        def enqueud_at(event)
+        def enqueued_at(event)
           Time.at(event.payload[:timestamp]).utc
         end
 
diff --git a/lib/active_job/performing.rb b/lib/active_job/performing.rb
new file mode 100644
index 0000000000..6c304a4bed
--- /dev/null
+++ b/lib/active_job/performing.rb
@@ -0,0 +1,14 @@
+require 'active_job/parameters'
+
+module ActiveJob
+  module Performing
+    def perform_with_deserialization(*serialized_args)
+      ActiveSupport::Notifications.instrument "perform.active_job", adapter: self.class.queue_adapter, job: self.class, args: serialized_args
+      perform *Parameters.deserialize(serialized_args)
+    end
+
+    def perform(*)
+      raise NotImplementedError
+    end
+  end
+end
diff --git a/lib/active_job/queue_adapters/backburner_adapter.rb b/lib/active_job/queue_adapters/backburner_adapter.rb
index 0d023d9ee7..12760316f0 100644
--- a/lib/active_job/queue_adapters/backburner_adapter.rb
+++ b/lib/active_job/queue_adapters/backburner_adapter.rb
@@ -11,12 +11,12 @@ module ActiveJob
         def queue_at(job, timestamp, *args)
           raise NotImplementedError
         end
-     end
+      end
 
       class JobWrapper
         class << self
           def perform(job_name, *args)
-            job_name.constantize.new.perform *Parameters.deserialize(args)
+            job_name.constantize.new.perform_with_deserialization *args
           end
         end
       end
diff --git a/lib/active_job/queue_adapters/delayed_job_adapter.rb b/lib/active_job/queue_adapters/delayed_job_adapter.rb
index 214733e3a6..db7b8a5528 100644
--- a/lib/active_job/queue_adapters/delayed_job_adapter.rb
+++ b/lib/active_job/queue_adapters/delayed_job_adapter.rb
@@ -15,7 +15,7 @@ module ActiveJob
 
       class JobWrapper
         def perform(job, *args)
-          job.new.perform *Parameters.deserialize(args)
+          job.new.perform_with_deserialization *args
         end
       end
     end
diff --git a/lib/active_job/queue_adapters/inline_adapter.rb b/lib/active_job/queue_adapters/inline_adapter.rb
index 4f703a6c3f..8f790943c4 100644
--- a/lib/active_job/queue_adapters/inline_adapter.rb
+++ b/lib/active_job/queue_adapters/inline_adapter.rb
@@ -3,7 +3,7 @@ module ActiveJob
     class InlineAdapter
       class << self
         def queue(job, *args)
-          job.new.perform *Parameters.deserialize(args)
+          job.new.perform_with_deserialization *args
         end
 
         def queue_at(job, timestamp, *args)
@@ -11,7 +11,7 @@ module ActiveJob
             begin
               interval = Time.now.to_f - timestamp
               sleep(interval) if interval > 0
-              job.new.perform *Parameters.deserialize(args)
+              job.new.perform_with_deserialization *args
             rescue => e
               ActiveJob::Base.logger.info "Error performing #{job}: #{e.message}"
             end
diff --git a/lib/active_job/queue_adapters/que_adapter.rb b/lib/active_job/queue_adapters/que_adapter.rb
index 30c23a35b9..30a69dbd64 100644
--- a/lib/active_job/queue_adapters/que_adapter.rb
+++ b/lib/active_job/queue_adapters/que_adapter.rb
@@ -15,7 +15,7 @@ module ActiveJob
 
       class JobWrapper < Que::Job
         def run(job, *args)
-          job.new.perform *Parameters.deserialize(args)
+          job.new.perform_with_deserialization *args
         end
       end
     end
diff --git a/lib/active_job/queue_adapters/queue_classic_adapter.rb b/lib/active_job/queue_adapters/queue_classic_adapter.rb
index d0e2e1aa22..762b0370d5 100644
--- a/lib/active_job/queue_adapters/queue_classic_adapter.rb
+++ b/lib/active_job/queue_adapters/queue_classic_adapter.rb
@@ -15,7 +15,7 @@ module ActiveJob
 
       class JobWrapper
         def self.perform(job, *args)
-          job.new.perform *Parameters.deserialize(args)
+          job.new.perform_with_deserialization *args
         end
       end
     end
diff --git a/lib/active_job/queue_adapters/resque_adapter.rb b/lib/active_job/queue_adapters/resque_adapter.rb
index 353b3ae690..f16d546c3c 100644
--- a/lib/active_job/queue_adapters/resque_adapter.rb
+++ b/lib/active_job/queue_adapters/resque_adapter.rb
@@ -19,7 +19,7 @@ module ActiveJob
       class JobWrapper
         class << self
           def perform(job_name, *args)
-            job_name.constantize.new.perform *Parameters.deserialize(args)
+            job_name.constantize.new.perform_with_deserialization *args
           end
         end
 
diff --git a/lib/active_job/queue_adapters/sidekiq_adapter.rb b/lib/active_job/queue_adapters/sidekiq_adapter.rb
index 9829049220..15cfd22bf5 100644
--- a/lib/active_job/queue_adapters/sidekiq_adapter.rb
+++ b/lib/active_job/queue_adapters/sidekiq_adapter.rb
@@ -26,7 +26,7 @@ module ActiveJob
         include Sidekiq::Worker
 
         def perform(job_name, *args)
-          job_name.constantize.new.perform *Parameters.deserialize(args)
+          job_name.constantize.new.perform_with_deserialization *args
         end
       end
     end
diff --git a/lib/active_job/queue_adapters/sneakers_adapter.rb b/lib/active_job/queue_adapters/sneakers_adapter.rb
index b299b25a96..91be8790dd 100644
--- a/lib/active_job/queue_adapters/sneakers_adapter.rb
+++ b/lib/active_job/queue_adapters/sneakers_adapter.rb
@@ -23,7 +23,7 @@ module ActiveJob
         include Sneakers::Worker
 
         def work(job, *args)
-          job.new.perform *Parameters.deserialize(args)
+          job.new.perform_with_deserialization *args
         end
       end
     end
diff --git a/lib/active_job/queue_adapters/sucker_punch_adapter.rb b/lib/active_job/queue_adapters/sucker_punch_adapter.rb
index 4163db1c35..22bf3ed799 100644
--- a/lib/active_job/queue_adapters/sucker_punch_adapter.rb
+++ b/lib/active_job/queue_adapters/sucker_punch_adapter.rb
@@ -17,7 +17,7 @@ module ActiveJob
         include SuckerPunch::Job
 
         def perform(job, *args)
-          job.new.perform *Parameters.deserialize(args)
+          job.new.perform_with_deserialization *args
         end
       end
     end