Best way to test async listeners of an event

[tgsoverly]

I haven't seen any clean way to pass an event into an async event handler. When you create an even and pass it to the ActiveJob for example it will give you serialization errors. I have pieced together the following from the source code:

event = Event.new(data: { test: "data" })
event.metadata[:timestamp] = Time.zone.now
event.metadata[:valid_at] = Time.zone.now
serialized_event = RubyEventStore::Mappers::Default.new.event_to_record(event).serialize(YAML)
Job.perform_now(serialized_event.to_h)

But that is pretty ugly to unit test a handler, any suggestions out there?

[fidel]

Hi @tgsoverly!

Thanks for bringing this up. There aren't many ways to do it better right now because of how ActiveJob::Base is implemented. However, for the tests relying on ActiveJob::TestHelper we recently found a way to do this more or less like that:

event = Event.new(data: { test: "data" })
event.metadata[:timestamp] = Time.zone.now
event.metadata[:valid_at] = Time.zone.now

perform_enqueued_jobs(only: Job) do
  Rails.configuration.event_store.publish(event)
end

Or make this a method and hide the implementation details:

class FancyTest < ActiveSupport::TestCase
  include ActiveJob::TestHelper

  def test_whatever
    event = Event.new(data: { test: "data" })
    event.metadata[:timestamp] = Time.zone.now
    event.metadata[:valid_at] = Time.zone.now

    handle_event_async(event, Job)

    assert sideffects_of_choice
  end

  private 

  def handle_event_async(event, async_handler_class)
    perform_enqueued_jobs(only: async_handler_class) do
      Rails.configuration.event_store.publish(event)
    end
  end

At least you let RailsEventstore::AsyncHandler do it's job and don't need to care about the serialization internals.

[mostlyobvious]

Long story short:

• when designing event handlers we've settled for anything that responds to #call: a lambda, Proc, Method, on any other object defining def call(event); end

• that didn't work for async handlers, either based on ActiveJob::Base or directly on Sidekiq::Worker

• a job has to be a named class (so that its name can be turned into string) and its payload has to serialized too (so that it can be transferred into Redis for workers to pick up)

• components responsible for triggering async handlers in RES are a dispatcher (when) and a scheduler (how)

• when job is taken from queue by a processor, its class is instantiated and its payload is passed to its main method, that is perform in ActiveJob

• to handle serialized event payload from RES, a job would have to do this:

  class JobToHandleSomeEvent < ActiveJob::Base
    def perform(payload)
      event = event_store.deserialize(serializer: serializer_used_by_scheduler, **payload.symbolize_keys)
      something_from_event_data = event.data.fetch(: something_from_event_data)
      ...
    end
  
    private
  
    def event_store
      Rails.configuration.event_store
    end
  end

• RES provides RailsEventStore::AsyncHandler helper module with its own perform method, which decorates the perform method in the job when prepended, allowing this:

  class JobToHandleSomeEvent < ActiveJob::Base
    prepend RailsEventStore::AsyncHandler
  
    def perform(event)
      something_from_event_data = event.data.fetch(:something_from_event_data)
      ...
    end
  end

• that helper module changes nothing on the input side of the job/handler, thus the errors when passing non-serialized event as an argument for perform_now in tests

That said, one pattern that I've practiced in the past was:

• having your own handler class, wrapping RES infrastructure, which invoked #call(event) in the end

  class MyAsyncHandler < ApplicationJob
    prepend RailsEventStore::AsyncHandler
  
    queue_as :default
  
    def perform(event)
      call(event)
    end
  
    protected
  
    def event_store
      Rails.configuration.event_store
    end
  end

• implementing async event handlers with their main method as #call(event)

  class JobToHandleSomeEvent < MyAsyncHandler
    def call(event)
      something_from_event_data = event.data.fetch(:something_from_event_data)
    end
  end

• assuming that perform is the interface for framework/infra code, while call is the interface for unit tests

  def test_handle_some_event
    event = MyEvent.new(data: { ... })
    JobToHandleSomeEvent.new.call(event)
    
    assert_on_sideffects_of_choice
  end

[mostlyobvious]

Sidenote: I made some attempts in the past to redesign async handlers and written my thoughts here — https://gist.github.com/pawelpacana/295337e92dd9edd2e443e6b31ac2683a