Subscriptions redesigned

rails_event_store/lib/rails_event_store/active_job_scheduler.rb

@@ -4,7 +4,8 @@
 
 module RailsEventStore
   class ActiveJobScheduler
-    def call(klass, serialized_event)
+    def call(subscription, serialized_event)
+      klass = subscription.subscriber
       klass.perform_later(serialized_event.to_h)
     end
 

rails_event_store/lib/rails_event_store/after_commit_async_dispatcher.rb

@@ -6,9 +6,9 @@ def initialize(scheduler:)
       @scheduler = scheduler
     end
 
-    def call(subscriber, _, serialized_event)
+    def call(subscription, _, serialized_event)
       run do
-        @scheduler.call(subscriber, serialized_event)
+        @scheduler.call(subscription.subscriber, serialized_event)
       end
     end
 

rails_event_store/spec/active_job_scheduler_spec.rb

@@ -54,7 +54,8 @@ module RailsEventStore
     describe "#call" do
       specify do
         scheduler = ActiveJobScheduler.new
-        scheduler.call(MyAsyncHandler, serialized_event)
+        subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+        scheduler.call(subscription, serialized_event)
 
         enqueued_jobs = ActiveJob::Base.queue_adapter.enqueued_jobs
         expect(enqueued_jobs.size).to eq(1)

rails_event_store/spec/after_commit_async_dispatcher_spec.rb

@@ -37,7 +37,8 @@ class DummyRecord < ActiveRecord::Base
 
     it "dispatch job immediately when no transaction is open" do
       expect_to_have_enqueued_job(MyAsyncHandler) do
-        dispatcher.call(MyAsyncHandler, event, serialized_event)
+        subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+        dispatcher.call(subscription, event, serialized_event)
       end
       expect(MyAsyncHandler.received).to be_nil
       MyAsyncHandler.perform_enqueued_jobs
@@ -48,7 +49,8 @@ class DummyRecord < ActiveRecord::Base
       expect_to_have_enqueued_job(MyAsyncHandler) do
         ActiveRecord::Base.transaction do
           expect_no_enqueued_job(MyAsyncHandler) do
-            dispatcher.call(MyAsyncHandler, event, serialized_event)
+            subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+            dispatcher.call(subscription, event, serialized_event)
           end
         end
       end
@@ -61,7 +63,8 @@ class DummyRecord < ActiveRecord::Base
       it "does not dispatch job" do
         expect_no_enqueued_job(MyAsyncHandler) do
           ActiveRecord::Base.transaction do
-            dispatcher.call(MyAsyncHandler, event, serialized_event)
+            subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+            dispatcher.call(subscription, event, serialized_event)
             raise ActiveRecord::Rollback
           end
         end
@@ -84,7 +87,8 @@ class DummyRecord < ActiveRecord::Base
         it "does not dispatch job" do
           expect_no_enqueued_job(MyAsyncHandler) do
             ActiveRecord::Base.transaction do
-              dispatcher.call(MyAsyncHandler, event, serialized_event)
+              subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+              dispatcher.call(subscription, event, serialized_event)
               raise ActiveRecord::Rollback
             end
           end
@@ -99,7 +103,8 @@ class DummyRecord < ActiveRecord::Base
         ActiveRecord::Base.transaction do
           expect_no_enqueued_job(MyAsyncHandler) do
             ActiveRecord::Base.transaction(requires_new: false) do
-              dispatcher.call(MyAsyncHandler, event, serialized_event)
+              subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+              dispatcher.call(subscription, event, serialized_event)
             end
           end
         end
@@ -114,7 +119,8 @@ class DummyRecord < ActiveRecord::Base
         ActiveRecord::Base.transaction do
           expect_no_enqueued_job(MyAsyncHandler) do
             ActiveRecord::Base.transaction(requires_new: true) do
-              dispatcher.call(MyAsyncHandler, event, serialized_event)
+              subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+              dispatcher.call(subscription, event, serialized_event)
             end
           end
         end
@@ -129,7 +135,8 @@ class DummyRecord < ActiveRecord::Base
         ActiveRecord::Base.transaction do
           expect_no_enqueued_job(MyAsyncHandler) do
             ActiveRecord::Base.transaction(requires_new: true) do
-              dispatcher.call(MyAsyncHandler, event, serialized_event)
+              subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+              dispatcher.call(subscription, event, serialized_event)
               raise ActiveRecord::Rollback
             end
           end
@@ -150,7 +157,8 @@ class DummyRecord < ActiveRecord::Base
             ActiveRecord::Base.transaction do
               DummyRecord.new.save!
               expect_no_enqueued_job(MyAsyncHandler) do
-                dispatcher.call(MyAsyncHandler, event, serialized_event)
+                subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+                dispatcher.call(subscription, event, serialized_event)
               end
             end
           rescue DummyError
@@ -181,7 +189,8 @@ class DummyRecord < ActiveRecord::Base
               ActiveRecord::Base.transaction do
                 DummyRecord.new.save!
                 expect_no_enqueued_job(MyAsyncHandler) do
-                  dispatcher.call(MyAsyncHandler, event, serialized_event)
+                  subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+                  dispatcher.call(subscription, event, serialized_event)
                 end
               end
             rescue DummyError
@@ -205,7 +214,8 @@ class DummyRecord < ActiveRecord::Base
 
       it "dispatches the job" do
         expect_to_have_enqueued_job(MyAsyncHandler) do
-          dispatcher.call(MyAsyncHandler, event, serialized_event)
+          subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+          dispatcher.call(subscription, event, serialized_event)
         end
       end
 
@@ -218,7 +228,8 @@ class DummyRecord < ActiveRecord::Base
           expect_no_enqueued_job(MyAsyncHandler) do
             ActiveRecord::Base.transaction do
               expect_no_enqueued_job(MyAsyncHandler) do
-                dispatcher.call(MyAsyncHandler, event, serialized_event)
+                subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+                dispatcher.call(subscription, event, serialized_event)
               end
             end
           end
@@ -234,7 +245,8 @@ class DummyRecord < ActiveRecord::Base
           expect_to_have_enqueued_job(MyAsyncHandler) do
             ActiveRecord::Base.transaction do
               expect_no_enqueued_job(MyAsyncHandler) do
-                dispatcher.call(MyAsyncHandler, event, serialized_event)
+                subscription = RubyEventStore::Subscription.new(MyAsyncHandler)
+                dispatcher.call(subscription, event, serialized_event)
               end
             end
           end

rails_event_store/spec/rails_event_store_spec.rb

@@ -51,7 +51,8 @@ def perform(_event)
   end
 
   class CustomSidekiqScheduler
-    def call(klass, serialized_event)
+    def call(subscription, serialized_event)
+      klass = subscription.subscriber
       klass.perform_async(serialized_event.to_h)
     end
 

ruby_event_store/lib/ruby_event_store.rb

@@ -1,6 +1,8 @@
 # frozen_string_literal: true
 
 require 'ruby_event_store/dispatcher'
+require 'ruby_event_store/subscription'
+require 'ruby_event_store/in_memory_subscriptions_store'
 require 'ruby_event_store/subscriptions'
 require 'ruby_event_store/broker'
 require 'ruby_event_store/in_memory_repository'

ruby_event_store/lib/ruby_event_store/client.rb

@@ -176,7 +176,7 @@ def call
         unsubs += add_thread_subscribers
         @block.call
       ensure
-        unsubs.each(&:call) if unsubs
+        unsubs.each(&:unsubscribe) if unsubs
       end
 
       private

ruby_event_store/lib/ruby_event_store/composed_dispatcher.rb

@@ -6,10 +6,10 @@ def initialize(*dispatchers)
       @dispatchers = dispatchers
     end
 
-    def call(subscriber, event, serialized_event)
+    def call(subscription, event, serialized_event)
       @dispatchers.each do |dispatcher|
-        if dispatcher.verify(subscriber)
-          dispatcher.call(subscriber, event, serialized_event)
+        if dispatcher.verify(subscription.subscriber)
+          dispatcher.call(subscription, event, serialized_event)
           break
         end
       end

ruby_event_store/lib/ruby_event_store/constants.rb

@@ -2,5 +2,6 @@
 
 module RubyEventStore
   GLOBAL_STREAM = Object.new
+  GLOBAL_SUBSCRIPTION = Object.new
   PAGE_SIZE = 100.freeze
 end

ruby_event_store/lib/ruby_event_store/dispatcher.rb

@@ -2,9 +2,8 @@
 
 module RubyEventStore
   class Dispatcher
-    def call(subscriber, event, _)
-      subscriber = subscriber.new if Class === subscriber
-      subscriber.call(event)
+    def call(subscription, event, _)
+      subscription.call(event)
     end
 
     def verify(subscriber)

ruby_event_store/lib/ruby_event_store/immediate_async_dispatcher.rb

@@ -6,8 +6,8 @@ def initialize(scheduler:)
       @scheduler = scheduler
     end
 
-    def call(subscriber, _, serialized_event)
-      @scheduler.call(subscriber, serialized_event)
+    def call(subscription, _, serialized_event)
+      @scheduler.call(subscription, serialized_event)
     end
 
     def verify(subscriber)

ruby_event_store/lib/ruby_event_store/in_memory_subscriptions_store.rb

@@ -0,0 +1,23 @@
+module RubyEventStore
+  class InMemorySubscriptionsStore
+    def initialize
+      @subscriptions = Hash.new {|hsh, key| hsh[key] = [] }
+    end
+
+    def add(subscription, type = GLOBAL_SUBSCRIPTION)
+      @subscriptions[type.to_s] << subscription
+    end
+
+    def delete(subscription, type = GLOBAL_SUBSCRIPTION)
+      @subscriptions.fetch(type.to_s).delete(subscription)
+    end
+
+    def all_for(event_type)
+      @subscriptions[event_type.to_s]
+    end
+
+    def value
+      self
+    end
+  end
+end

ruby_event_store/lib/ruby_event_store/instrumented_dispatcher.rb

@@ -7,9 +7,9 @@ def initialize(dispatcher, instrumentation)
       @instrumentation = instrumentation
     end
 
-    def call(subscriber, event, serialized_event)
-      instrumentation.instrument("call.dispatcher.rails_event_store", event: event, subscriber: subscriber) do
-        dispatcher.call(subscriber, event, serialized_event)
+    def call(subscription, event, serialized_event)
+      instrumentation.instrument("call.dispatcher.rails_event_store", event: event, subscription: subscription) do
+        dispatcher.call(subscription, event, serialized_event)
       end
     end
 

ruby_event_store/lib/ruby_event_store/spec/subscriptions_lint.rb

@@ -25,10 +25,9 @@ def call(event)
     subscriptions.add_subscription(handler, [Test1DomainEvent, Test3DomainEvent])
     subscriptions.add_subscription(another_handler, [Test2DomainEvent])
     subscriptions.add_global_subscription(global_handler)
-
-    expect(subscriptions.all_for('Test1DomainEvent')).to eq([handler, global_handler])
-    expect(subscriptions.all_for('Test2DomainEvent')).to eq([another_handler, global_handler])
-    expect(subscriptions.all_for('Test3DomainEvent')).to eq([handler, global_handler])
+    expect(subscriptions.all_for('Test1DomainEvent').map(&:subscriber)).to eq([handler, global_handler])
+    expect(subscriptions.all_for('Test2DomainEvent').map(&:subscriber)).to eq([another_handler, global_handler])
+    expect(subscriptions.all_for('Test3DomainEvent').map(&:subscriber)).to eq([handler, global_handler])
   end
 
   it 'returns subscribed thread handlers' do
@@ -40,9 +39,9 @@ def call(event)
     subscriptions.add_thread_subscription(another_handler, [Test2DomainEvent])
     subscriptions.add_thread_global_subscription(global_handler)
 
-    expect(subscriptions.all_for('Test1DomainEvent')).to eq([global_handler, handler])
-    expect(subscriptions.all_for('Test2DomainEvent')).to eq([global_handler, another_handler])
-    expect(subscriptions.all_for('Test3DomainEvent')).to eq([global_handler, handler])
+    expect(subscriptions.all_for('Test1DomainEvent').map(&:subscriber)).to eq([global_handler, handler])
+    expect(subscriptions.all_for('Test2DomainEvent').map(&:subscriber)).to eq([global_handler, another_handler])
+    expect(subscriptions.all_for('Test3DomainEvent').map(&:subscriber)).to eq([global_handler, handler])
   end
 
   it 'returns lambda as an output of global subscribe methods' do
@@ -60,43 +59,43 @@ def call(event)
   it 'revokes global subscription' do
     handler   = TestHandler.new
 
-    revoke    = subscriptions.add_global_subscription(handler)
-    expect(subscriptions.all_for('Test1DomainEvent')).to eq([handler])
-    expect(subscriptions.all_for('Test2DomainEvent')).to eq([handler])
-    revoke.()
+    subscription = subscriptions.add_global_subscription(handler)
+    expect(subscriptions.all_for('Test1DomainEvent').map(&:subscriber)).to eq([handler])
+    expect(subscriptions.all_for('Test2DomainEvent').map(&:subscriber)).to eq([handler])
+    subscription.unsubscribe
     expect(subscriptions.all_for('Test1DomainEvent')).to eq([])
     expect(subscriptions.all_for('Test2DomainEvent')).to eq([])
   end
 
   it 'revokes subscription' do
     handler   = TestHandler.new
 
-    revoke    = subscriptions.add_subscription(handler, [Test1DomainEvent, Test2DomainEvent])
-    expect(subscriptions.all_for('Test1DomainEvent')).to eq([handler])
-    expect(subscriptions.all_for('Test2DomainEvent')).to eq([handler])
-    revoke.()
+    subscription = subscriptions.add_subscription(handler, [Test1DomainEvent, Test2DomainEvent])
+    expect(subscriptions.all_for('Test1DomainEvent').map(&:subscriber)).to eq([handler])
+    expect(subscriptions.all_for('Test2DomainEvent').map(&:subscriber)).to eq([handler])
+    subscription.unsubscribe
     expect(subscriptions.all_for('Test1DomainEvent')).to eq([])
     expect(subscriptions.all_for('Test2DomainEvent')).to eq([])
   end
 
   it 'revokes thread global subscription' do
     handler   = TestHandler.new
 
-    revoke    = subscriptions.add_thread_global_subscription(handler)
-    expect(subscriptions.all_for('Test1DomainEvent')).to eq([handler])
-    expect(subscriptions.all_for('Test2DomainEvent')).to eq([handler])
-    revoke.()
+    subscription = subscriptions.add_thread_global_subscription(handler)
+    expect(subscriptions.all_for('Test1DomainEvent').map(&:subscriber)).to eq([handler])
+    expect(subscriptions.all_for('Test2DomainEvent').map(&:subscriber)).to eq([handler])
+    subscription.unsubscribe
     expect(subscriptions.all_for('Test1DomainEvent')).to eq([])
     expect(subscriptions.all_for('Test2DomainEvent')).to eq([])
   end
 
   it 'revokes thread subscription' do
     handler           = TestHandler.new
 
-    revoke    = subscriptions.add_thread_subscription(handler, [Test1DomainEvent, Test2DomainEvent])
-    expect(subscriptions.all_for('Test1DomainEvent')).to eq([handler])
-    expect(subscriptions.all_for('Test2DomainEvent')).to eq([handler])
-    revoke.()
+    subscription = subscriptions.add_thread_subscription(handler, [Test1DomainEvent, Test2DomainEvent])
+    expect(subscriptions.all_for('Test1DomainEvent').map(&:subscriber)).to eq([handler])
+    expect(subscriptions.all_for('Test2DomainEvent').map(&:subscriber)).to eq([handler])
+    subscription.unsubscribe
     expect(subscriptions.all_for('Test1DomainEvent')).to eq([])
     expect(subscriptions.all_for('Test2DomainEvent')).to eq([])
   end
@@ -106,6 +105,6 @@ def call(event)
     subscriptions.add_subscription(handler, ["Test1DomainEvent"])
     subscriptions.add_thread_subscription(handler, ["Test1DomainEvent"])
 
-    expect(subscriptions.all_for('Test1DomainEvent')).to eq([handler, handler])
+    expect(subscriptions.all_for('Test1DomainEvent').map(&:subscriber)).to eq([handler, handler])
   end
 end