lib/bluepill/trigger.rb in evented_bluepill-0.0.46 vs lib/bluepill/trigger.rb in evented_bluepill-0.0.47

@@ -1,70 +1,72 @@
+# -*- encoding: utf-8 -*-
+
+require 'active_support'
+
 module Bluepill
+  class TriggerTimer < Coolio::TimerWatcher
+    def initialize(trigger, event, delay)
+      @trigger = trigger
+      @event = event
+      super(delay, false)
+    end
+
+    def on_timer
+      begin
+        @trigger.logger.info("Retrying from flapping")
+        @trigger.dispatch!(@event)
+        @trigger.remove_timer(self)
+      rescue StandardError => e
+        @trigger.logger.err(e)
+        @trigger.logger.err(e.backtrace.join("\n"))
+      end
+    end
+  end
+
   class Trigger
     @implementations = {}
     def self.inherited(klass)
       @implementations[klass.name.split('::').last.underscore.to_sym] = klass
     end
-    
+
     def self.[](name)
       @implementations[name]
     end
 
-    attr_accessor :process, :logger, :mutex, :scheduled_events
-    
+    attr_accessor :process, :logger, :scheduled_events
+
     def initialize(process, options = {})
       self.process = process
       self.logger = options[:logger]
-      self.mutex = Mutex.new
       self.scheduled_events = []
     end
-    
+
     def reset!
       self.cancel_all_events
     end
-    
+
     def notify(transition)
       raise "Implement in subclass"
     end
-    
+
     def dispatch!(event)
       self.process.dispatch!(event, self.class.name.split("::").last)
     end
-    
-    def deep_copy
-      # TODO: This is a kludge. Ideally, process templates 
-      # would be facotries, and not a template object.
-      mutex, @mutex = @mutex, nil
-      clone = Marshal.load(Marshal.dump(self))
-      clone.instance_variable_set("@mutex", Monitor.new)
-      @mutex = mutex
-      clone
+
+    def remove_timer(timer)
+      self.scheduled_events.delete(timer)
     end
-    
+
     def schedule_event(event, delay)
-      # TODO: maybe wrap this in a ScheduledEvent class with methods like cancel
-      thread = Thread.new(self) do |trigger|
-        begin
-          sleep delay.to_f
-          trigger.logger.info("Retrying from flapping")
-          trigger.dispatch!(event)
-          trigger.mutex.synchronize do
-            trigger.scheduled_events.delete_if { |_, thread| thread == Thread.current }
-          end
-        rescue StandardError => e
-          trigger.logger.err(e)
-          trigger.logger.err(e.backtrace.join("\n"))
-        end
-      end
+      timer = Bluepill::TriggerTimer.new(self, event, delay)
+      self.scheduled_events << timer
       
-      self.scheduled_events.push([event, thread])
+      Bluepill::Event.attach(timer)
     end
-    
+
     def cancel_all_events
       self.logger.info "Canceling all scheduled events"
-      self.mutex.synchronize do
-        self.scheduled_events.each {|_, thread| thread.kill}
-      end
+      self.scheduled_events.each {|e| e.detach }
+      self.scheduled_events.clear
     end
-    
   end
 end
\ No newline at end of file