Subscriptions redesigned

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/global_subscription'
 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/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/global_subscription.rb

@@ -0,0 +1,59 @@
+module RubyEventStore
+  class GlobalSubscription
+    def initialize(subscriber, store)
+      @subscriber = subscriber
+      @store = store
+      @store.add(self)
+    end
+
+    def call(event)
+      (Class === subscriber ? subscriber.new : subscriber).call(event)
+    end
+
+    def unsubscribe
+      @store.delete(self)
+    end
+
+    def inspect
+      <<~EOS.strip
+          #<#{self.class}:0x#{__id__.to_s(16)}>
+            - subscriber: #{subscriber.inspect}
+      EOS
+    end
+
+    # Two subscriptions are equal if:
+    # * they are of the same class
+    # * have identical subscriber (verified with eql? method)
+    #
+    # @param other [GlobalSubscription, Object] object to compare
+    #
+    # Event equality ignores metadata!
+    # @return [TrueClass, FalseClass]
+    def ==(other)
+      other.instance_of?(self.class) &&
+        other.subscriber.eql?(subscriber)
+    end
+
+    # @private
+    BIG_VALUE = 0b101000101111010111101101010011000101000100000000011011011100110
+
+    # Generates a Fixnum hash value for this object. This function
+    # have the property that a.eql?(b) implies a.hash == b.hash.
+    #
+    # The hash value is used along with eql? by the Hash class to
+    # determine if two objects reference the same hash key.
+    #
+    # This hash is based on
+    # * class
+    # * subscriber object_id
+    def hash
+      # We don't use metadata because == does not use metadata
+      [
+        self.class,
+        subscriber.object_id,
+      ].hash ^ BIG_VALUE
+    end
+
+    attr_reader :subscriber
+  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

ruby_event_store/lib/ruby_event_store/subscription.rb

@@ -0,0 +1,65 @@
+module RubyEventStore
+  class Subscription
+    def initialize(subscriber, event_types, store)
+      @subscriber = subscriber
+      @store = store
+      @event_types = event_types
+      event_types.each{ |type| @store.add(type, self) }
+    end
+
+    def call(event)
+      (Class === subscriber ? subscriber.new : subscriber).call(event)
+    end
+
+    def unsubscribe
+      event_types.each{ |type| @store.delete(type, self) }
+    end
+
+    def inspect
+      <<~EOS.strip
+          #<#{self.class}:0x#{__id__.to_s(16)}>
+            - event types: #{event_types.inspect}
+            - subscriber: #{subscriber.inspect}
+      EOS
+    end
+
+    # Two subscriptions are equal if:
+    # * they are of the same class
+    # * have identical event types
+    # * have identical subscriber (verified with eql? method)
+    #
+    # @param other [Subscription, Object] object to compare
+    #
+    # Event equality ignores metadata!
+    # @return [TrueClass, FalseClass]
+    def ==(other)
+      other.instance_of?(self.class) &&
+        other.event_types.eql?(event_types) &&
+        other.subscriber.eql?(subscriber)
+    end
+
+    # @private
+    BIG_VALUE = 0b11010000100100100101110000000010011110000110101011010001001110
+
+    # Generates a Fixnum hash value for this object. This function
+    # have the property that a.eql?(b) implies a.hash == b.hash.
+    #
+    # The hash value is used along with eql? by the Hash class to
+    # determine if two objects reference the same hash key.
+    #
+    # This hash is based on
+    # * class
+    # * event types
+    # * subscriber objecy id
+    def hash
+      # We don't use metadata because == does not use metadata
+      [
+        self.class,
+        event_types,
+        subscriber.object_id
+      ].hash ^ BIG_VALUE
+    end
+
+    attr_reader :subscriber, :event_types
+  end
+end

ruby_event_store/lib/ruby_event_store/subscriptions.rb

@@ -4,10 +4,10 @@
 
 module RubyEventStore
   class Subscriptions
-    def initialize
-      @local  = LocalSubscriptions.new
-      @global = GlobalSubscriptions.new
-      @thread  = ThreadSubscriptions.new
+    def initialize(local_store: Store, global_store: GlobalStore)
+      @local  = LocalSubscriptions.new(local_store.new)
+      @global = GlobalSubscriptions.new(global_store.new)
+      @thread = ThreadSubscriptions.new(local_store, global_store)
     end
 
     def add_subscription(subscriber, event_types)
@@ -33,77 +33,95 @@ def all_for(event_type)
     private
     attr_reader :local, :global, :thread
 
-    class ThreadSubscriptions
+    class Store
       def initialize
-        @local  = ThreadLocalSubscriptions.new
-        @global = ThreadGlobalSubscriptions.new
-      end
-      attr_reader :local, :global
-
-      def all_for(event_type)
-        [global, local].map{|r| r.all_for(event_type)}.reduce(&:+)
+        @subscriptions = Hash.new {|hsh, key| hsh[key] = [] }
       end
-    end
 
-    class LocalSubscriptions
-      def initialize
-        @subscriptions = Hash.new {|hsh, key| hsh[key] = [] }
+      def add(type, subscription)
+        @subscriptions[type.to_s] << subscription
       end
 
-      def add(subscription, event_types)
-        event_types.each{ |type| @subscriptions[type.to_s] << subscription }
-        ->() {event_types.each{ |type| @subscriptions.fetch(type.to_s).delete(subscription) } }
+      def delete(type, subscription)
+        @subscriptions.fetch(type.to_s).delete(subscription)
       end
 
       def all_for(event_type)
         @subscriptions[event_type]
       end
+
+      def value
+        self
+      end
     end
 
-    class GlobalSubscriptions
+    class GlobalStore
       def initialize
         @subscriptions = []
       end
 
       def add(subscription)
         @subscriptions << subscription
-        ->() { @subscriptions.delete(subscription) }
       end
 
-      def all_for(_event_type)
+      def delete(subscription)
+        @subscriptions.delete(subscription)
+      end
+
+      def all
         @subscriptions
       end
+
+      def value
+        self
+      end
     end
 
-    class ThreadLocalSubscriptions
-      def initialize
-        @subscriptions = Concurrent::ThreadLocalVar.new do
-          Hash.new {|hsh, key| hsh[key] = [] }
-        end
+    class ThreadSubscriptions
+      def initialize(local_store, global_store)
+        @local  = LocalSubscriptions.new(build_store(local_store))
+        @global = GlobalSubscriptions.new(build_store(global_store))
+      end
+      attr_reader :local, :global
+
+      def all_for(event_type)
+        [global, local].map{|r| r.all_for(event_type)}.reduce(&:+)
+      end
+
+      private
+
+      def build_store(klass)
+        var = Concurrent::ThreadLocalVar.new(klass.new)
+        var.value = klass.new
+        var
+      end
+    end
+
+    class LocalSubscriptions
+      def initialize(store)
+        @store = store
       end
 
       def add(subscription, event_types)
-        event_types.each{ |type| @subscriptions.value[type.to_s] << subscription }
-        ->() {event_types.each{ |type| @subscriptions.value.fetch(type.to_s).delete(subscription) } }
+        Subscription.new(subscription, event_types, @store.value)
       end
 
       def all_for(event_type)
-        @subscriptions.value[event_type]
+        @store.value.all_for(event_type)
       end
     end
 
-    class ThreadGlobalSubscriptions
-      def initialize
-        @subscriptions = Concurrent::ThreadLocalVar.new([])
+    class GlobalSubscriptions
+      def initialize(store)
+        @store = store
       end
 
       def add(subscription)
-        @subscriptions.value += [subscription]
-        ->() { @subscriptions.value -= [subscription] }
+        GlobalSubscription.new(subscription, @store.value)
       end
 
       def all_for(_event_type)
-        @subscriptions.value
+        @store.value.all
       end
     end
   end