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

@@ -1,3 +1,5 @@
+# frozen_string_literal: true
+
 require 'spec_helper'
 require 'ruby_event_store/spec/scheduler_lint'
 
@@ -54,7 +56,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

@@ -1,3 +1,5 @@
+# frozen_string_literal: true
+
 require 'spec_helper'
 require 'ruby_event_store/spec/dispatcher_lint'
 require 'ruby_event_store/spec/scheduler_lint'
@@ -37,7 +39,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 +51,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 +65,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 +89,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 +105,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 +121,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 +137,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 +159,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 +191,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 +216,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 +230,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 +247,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

@@ -1,3 +1,5 @@
+# frozen_string_literal: true
+
 require 'spec_helper'
 require 'action_controller/railtie'
 
@@ -51,7 +53,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/Makefile

@@ -5,6 +5,7 @@ IGNORE      = RubyEventStore::InMemoryRepository\#append_with_synchronize \
               RubyEventStore::Client::Within\#add_thread_subscribers \
               RubyEventStore::Client::Within\#add_thread_global_subscribers \
               RubyEventStore::Client::Within\#call \
+              RubyEventStore::Subscriptions::ThreadSubscriptions\#build_store \
               RubyEventStore::Mappers::InMemoryEncryptionKeyRepository\#prepare_encrypt \
               RubyEventStore::Mappers::EncryptionKey\#prepare_encrypt \
               RubyEventStore::Mappers::EncryptionKey\#prepare_decrypt \

ruby_event_store/lib/ruby_event_store.rb

@@ -1,6 +1,9 @@
 # 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/instrumented_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,45 @@
+# frozen_string_literal: true
+
+module RubyEventStore
+  # In memory subscriptions store
+  class InMemorySubscriptionsStore
+    # Instantiates a new in memory subscriptions store
+    #
+    # @return [InMemorySubscriptionsStore]
+    def initialize
+      @subscriptions = Hash.new { |hsh, key| hsh[key] = [] }
+    end
+
+    # Stores subscription in the store
+    # @param subscription [Subscription] subscription to store
+    def add(subscription)
+      subscription.subscribed_for.each do |type|
+        @subscriptions[type.to_s] << subscription
+      end
+    end
+
+    # Removes subscription from the store
+    # @param subscription [Subscription] subscription to remove
+    def delete(subscription)
+      subscription.subscribed_for.each do |type|
+        @subscriptions.fetch(type.to_s).delete(subscription)
+      end
+    end
+
+    # Gets all subscriptions stored for given event type
+    # @param type [String, Class, GLOBAL_SUBSCRIPTION] a type of [Event]
+    #        or global subscription for which subscriptions should be returned
+    #
+    # @return [Array<Subscription>]
+    def all_for(type)
+      @subscriptions[type.to_s]
+    end
+
+    # Gets all subscriptions stored
+    #
+    # @return [Array<Subscription>]
+    def all
+      @subscriptions.values.flatten.uniq
+    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/instrumented_subscriptions_store.rb

@@ -0,0 +1,53 @@
+# frozen_string_literal: true
+
+module RubyEventStore
+  # Instrumented subscriptions store
+  class InstrumentedSubscriptionsStore
+    # Instantiates a new in memory subscriptions store
+    #
+    # @return [InstrumentedSubscriptionsStore]
+    def initialize(store, instrumentation)
+      @store = store
+      @instrumentation = instrumentation
+    end
+
+    # Stores subscription in the store
+    # @param subscription [Subscription] subscription to store
+    #
+    # @return [self]
+    def add(subscription)
+      instrumentation.instrument("add.subscription_store.rails_event_store", subscription: subscription) do
+        store.add(subscription)
+      end
+    end
+
+    # Removes subscription from the store
+    # @param subscription [Subscription] subscription to remove
+    #
+    # @return [self]
+    def delete(subscription)
+      instrumentation.instrument("delete.subscription_store.rails_event_store", subscription: subscription) do
+        store.delete(subscription)
+      end
+    end
+
+    # Gets all subscriptions stored for given event type
+    # @param type [String, Class, GLOBAL_SUBSCRIPTION] a type of [Event]
+    #        or global subscription for which subscriptions should be returned
+    #
+    # @return [Array<Subscription>]
+    def all_for(type)
+      store.all_for(type)
+    end
+
+    # Gets all subscriptions stored
+    #
+    # @return [Array<Subscription>]
+    def all
+      store.all
+    end
+
+    private
+    attr_reader :instrumentation, :store
+  end
+end