class ActiveEventStore::SubscriberJob

Base job for async subscribers

Attributes

subscriber[RW]

Public Class Methods

for(callable) click to toggle source
# File lib/active_event_store/subscriber_job.rb, line 31
def for(callable)
  raise ArgumentError, "Async subscriber must be a module/class" unless callable.is_a?(Module)

  callable.const_defined?("SubscriberJob", false) ?
    callable.const_get("SubscriberJob", false) :
    nil
end
from(callable) click to toggle source
# File lib/active_event_store/subscriber_job.rb, line 9
def from(callable)
  if callable.is_a?(Proc) || callable.name.nil?
    raise ArgumentError, "Anonymous subscribers (blocks/procs/lambdas or anonymous modules) " \
                          "could not be asynchronous (use sync: true)"
  end

  raise ArgumentError, "Async subscriber must be a module/class, not instance" unless callable.is_a?(Module)

  if callable.const_defined?("SubscriberJob", false)
    callable.const_get("SubscriberJob", false)
  else
    callable.const_set(
      "SubscriberJob",
      Class.new(self).tap do |job|
        queue_as ActiveEventStore.config.job_queue_name

        job.subscriber = callable
      end
    )
  end
end

Public Instance Methods

perform(payload) click to toggle source
# File lib/active_event_store/subscriber_job.rb, line 40
def perform(payload)
  event = event_store.deserialize(payload)

  event_store.with_metadata(**event.metadata.to_h) do
    subscriber.call(event)
  end
end

Private Instance Methods

event_store() click to toggle source
# File lib/active_event_store/subscriber_job.rb, line 54
def event_store
  ActiveEventStore.event_store
end
subscriber() click to toggle source
# File lib/active_event_store/subscriber_job.rb, line 50
def subscriber
  self.class.subscriber
end