Contents

module Reactor::Subscribable
  extend ActiveSupport::Concern

  module ClassMethods
    def on_event(*args, &block)
      options = args.extract_options!
      event, method = args
      (Reactor::SUBSCRIBERS[event.to_s] ||= []).push(StaticSubscriberFactory.create(event, method, {source: self}.merge(options), &block))
    end
  end

  class StaticSubscriberFactory

    def self.create(event, method = nil, options = {}, &block)
      handler_class_prefix = event == '*' ? 'Wildcard': event.to_s.camelize
      i = 0
      begin
        new_class = "#{handler_class_prefix}Handler#{i}"
        i+= 1
      end while Reactor::StaticSubscribers.const_defined?(new_class)

      klass = Class.new do
        include Sidekiq::Worker

        class_attribute :method, :delay, :source, :in_memory, :dont_perform

        def perform(data)
          return :__perform_aborted__ if dont_perform && !Reactor::TEST_MODE_SUBSCRIBERS.include?(source)
          event = Reactor::Event.new(data)
          if method.is_a?(Symbol)
            ActiveSupport::Deprecation.silence do
              source.delay_for(delay).send(method, event)
            end
          else
            method.call(event)
          end
        end

        def self.perform_where_needed(data)
          if in_memory
            new.perform(data)
          else
            perform_async(data)
          end
        end
      end

      Reactor::StaticSubscribers.const_set(new_class, klass)

      klass.tap do |k|
        k.method = method || block
        k.delay = options[:delay] || 0
        k.source = options[:source]
        k.in_memory = options[:in_memory]
        k.dont_perform = Reactor.test_mode?
      end
    end
  end
end

Version data entries

4 entries across 4 versions & 1 rubygems

Version	Path
reactor-0.10.0	lib/reactor/models/concerns/subscribable.rb
reactor-0.9.9	lib/reactor/models/concerns/subscribable.rb
reactor-0.9.8	lib/reactor/models/concerns/subscribable.rb
reactor-0.9.7	lib/reactor/models/concerns/subscribable.rb