Refactored event publishers to have single publish method w/type arg.

Author: lwoodson

README.md

@@ -62,7 +62,7 @@ You can implement your own publishers as long as they implement event handling
 methods as follows:
 
 ```ruby
-def event_type(timestamp, queue, metadata, klass, *args)
+def publish(event_type, timestamp, queue, metadata, klass, *args)
   ...
 end
 ```

TODO

@@ -1,3 +1,4 @@
-* Do not symbolize the item hash; Use string keys
-* Make hostname/process to be memoized.
-* Check return behavior of block.
+* Modify the publish signature to accept a hash instead of all the params.
+* Add file-like streams that wrap TCP/IP and UDP sockets?  (not needed for our
+  case).  Would allow use of StreamPublisher to communicate over those
+  protocols

lib/resque/plugins/clues/event_publisher.rb

@@ -24,11 +24,10 @@ def initialize(stream)
           @stream = stream
         end
 
-        EVENT_TYPES.each do |event_type|
-          define_method(event_type) do |timestamp, queue, metadata, klass, *args|
-            event = CLUES.event_marshaller.call(event_type, timestamp, queue, metadata, klass, args)
-            stream.write(event)
-          end
+        # Publishes an event to the stream.
+        def publish(event_type, timestamp, queue, metadata, klass, *args)
+          event = CLUES.event_marshaller.call(event_type, timestamp, queue, metadata, klass, args)
+          stream.write(event)
         end
       end
 
@@ -58,11 +57,10 @@ def initialize(log_path, formatter=nil)
           @logger.formatter = formatter || lambda {|severity, time, program, msg| msg}
         end
 
-        EVENT_TYPES.each do |event_type|
-          define_method(event_type) do |timestamp, queue, metadata, klass, *args|
-            logger.info(CLUES.event_marshaller.call(
-              event_type, timestamp, queue, metadata, klass, args))
-          end
+        # Publishes an event to the log.
+        def publish(event_type, timestamp, queue, metadata, klass, *args)
+          logger.info(CLUES.event_marshaller.call(
+            event_type, timestamp, queue, metadata, klass, args))
         end
       end
 
@@ -74,13 +72,13 @@ def initialize
           super([])
         end
 
-        EVENT_TYPES.each do |event_type|
-          define_method(event_type) do |timestamp, queue, metadata, klass, *args|
-            each do |child|
-              child.send(
-                event_type, timestamp, queue, metadata, klass, *args
-              ) rescue error(event_type, child)
-            end
+        # Invokes publish on each child publisher for them to publish the event
+        # in their own way.
+        def publish(event_type, timestamp, queue, metadata, klass, *args)
+          each do |child|
+            child.publish(
+              event_type, timestamp, queue, metadata, klass, *args
+            ) rescue error(event_type, child)
           end
         end
 

lib/resque/plugins/clues/job_extension.rb

@@ -29,11 +29,11 @@ def self.define_perform(klass) # :doc:
           klass.send(:define_method, :perform) do
             return _base_perform unless CLUES.configured?
             item = CLUES.prepare(payload)
-            CLUES.event_publisher.perform_started(CLUES.now, queue, item['clues_metadata'], item['class'], *item['args'])
+            CLUES.event_publisher.publish(:perform_started, CLUES.now, queue, item['clues_metadata'], item['class'], *item['args'])
             @perform_started = Time.now
             _base_perform.tap do
               item['clues_metadata']['time_to_perform'] = CLUES.time_delta_since(@perform_started)
-              CLUES.event_publisher.perform_finished(CLUES.now, queue, item['clues_metadata'], item['class'], *item['args'])
+              CLUES.event_publisher.publish(:perform_finished, CLUES.now, queue, item['clues_metadata'], item['class'], *item['args'])
             end
           end
         end
@@ -53,7 +53,7 @@ def self.define_failed(klass) # :doc:
                 item['clues_metadata']['exception'] = exception.class
                 item['clues_metadata']['message'] = exception.message
                 item['clues_metadata']['backtrace'] = exception.backtrace
-                CLUES.event_publisher.failed(CLUES.now, queue, item['clues_metadata'], item['class'], *item['args'])
+                CLUES.event_publisher.publish(:failed, CLUES.now, queue, item['clues_metadata'], item['class'], *item['args'])
               end
             end
           end

lib/resque/plugins/clues/queue_extension.rb

@@ -43,7 +43,7 @@ def push(queue, orig)
           if Resque::Plugins::Clues.item_preprocessor
             Resque::Plugins::Clues.item_preprocessor.call(queue, item)
           end
-          CLUES.event_publisher.enqueued(CLUES.now, queue, item['clues_metadata'], item['class'], *item['args'])
+          CLUES.event_publisher.publish(:enqueued, CLUES.now, queue, item['clues_metadata'], item['class'], *item['args'])
           _base_push(queue, item)
         end
 
@@ -58,7 +58,7 @@ def pop(queue)
               return orig unless CLUES.clues_configured?
               item = CLUES.prepare(orig) do |item|
                 item['clues_metadata']['time_in_queue'] = CLUES.time_delta_since(item['clues_metadata']['enqueued_time'])
-                CLUES.event_publisher.dequeued(CLUES.now, queue, item['clues_metadata'], item['class'], *item['args'])
+                CLUES.event_publisher.publish(:dequeued, CLUES.now, queue, item['clues_metadata'], item['class'], *item['args'])
               end
             end
           end

spec/event_publisher_spec.rb

@@ -12,7 +12,7 @@
   end
 
   def publish_event_type(type)
-    @publisher.send(type, @current_time, :test_queue, {}, "FooBar", "a", "b")
+    @publisher.publish(type, @current_time, :test_queue, {}, "FooBar", "a", "b")
   end
 
   describe Resque::Plugins::Clues::StreamPublisher do
@@ -145,8 +145,8 @@ def last_event
 
     def verify_event_delegated_to_children(event_type)
       @publisher.each do |child|
-        child.should_receive(event_type).with(
-          @current_time, :test_queue, {}, "FooBar", "a", "b")
+        child.should_receive(:publish).with(
+          event_type, @current_time, :test_queue, {}, "FooBar", "a", "b")
       end
     end
 

spec/job_extension_spec.rb

@@ -1,4 +1,5 @@
 require 'spec_helper'
+require 'set'
 
 describe Resque::Plugins::Clues::JobExtension do
   def base_item(overrides={})
@@ -31,40 +32,48 @@ def base_item(overrides={})
   end
 
   context "with clues configured" do
-    def publishes(evt_type)
-      Resque::Plugins::Clues.event_publisher.should_receive(evt_type)
-      Resque::Plugins::Clues.event_publisher.stub(evt_type) do |time, queue, metadata, klass, *args|
+    def publishes(opts={})
+      opts.keys.each{|key| @events_not_received << key}
+      @events_not_received.size.should_not == 0
+      Resque::Plugins::Clues.event_publisher.should_receive(:publish).at_least(:once)
+      Resque::Plugins::Clues.event_publisher.stub(:publish) do |type, time, queue, metadata, klass, *args|
+        @events_not_received.delete(type)
         time.nil?.should == false
         queue.should == :test_queue
         klass.should == 'TestWorker'
         args.should == [1,2]
         metadata['hostname'].should == `hostname`.strip
         metadata['process'].should == $$
-        yield(metadata) if block_given?
+        opts[type].call(metadata) if opts[type]
       end
     end
 
     before do
       Resque::Plugins::Clues.event_publisher = Resque::Plugins::Clues::StandardOutPublisher.new
+      @events_not_received = Set.new([])
+    end
+
+    after do
+      @events_not_received.size.should == 0 
     end
 
     describe "#perform" do
       it "should publish a perform_started event" do
-        publishes(:perform_started)
+        publishes perform_started: nil
         @job.perform
       end
 
       it "should publish a perform_finished event that includes the time_to_perform" do
-        publishes(:perform_finished) do |metadata|
+        publishes(perform_finished: lambda do |metadata|
           metadata['time_to_perform'].nil?.should == false
-        end
+        end)
         @job.perform
       end
     end
 
     describe "#fail" do
       it "should publish a perform_failed event" do
-        publishes(:failed)
+        publishes failed: nil
         @job.fail(Exception.new)
       end
 
@@ -75,33 +84,33 @@ def publishes(evt_type)
 
       context "includes metadata in the perform_failed event that should" do
         it "should include the time_to_perform" do
-          publishes(:failed) do |metadata|
+          publishes(failed: lambda do |metadata|
             metadata['time_to_perform'].nil?.should == false
-          end
+          end)
           @job.fail(Exception.new)
         end
 
         it "should include the exception class" do
-          publishes(:failed) do |metadata|
+          publishes(failed: lambda do |metadata|
             metadata['exception'].should == Exception
-          end
+          end)
           @job.fail(Exception.new)
         end
 
         it "should include the exception message" do
-          publishes(:failed) do |metadata|
+          publishes(failed: lambda do |metadata|
             metadata['message'].should == 'test'
-          end
+          end)
           @job.fail(Exception.new('test'))
         end
 
         it "should include the exception backtrace" do
           begin
             raise 'test'
           rescue => e
-            publishes(:failed) do |metadata|
+            publishes(failed: lambda do |metadata|
               metadata['backtrace'].nil?.should == false
-            end
+            end)
             @job.fail(e)
           end
         end