Add instrumentation for repositories

rails_event_store/lib/rails_event_store/client.rb

@@ -8,7 +8,7 @@ def initialize(repository: RailsEventStoreActiveRecord::EventRepository.new,
                    dispatcher: ActiveJobDispatcher.new,
                    request_metadata: default_request_metadata,
                    page_size: PAGE_SIZE)
-      super(repository: repository,
+      super(repository: RubyEventStore::InstrumentedRepository.new(repository, ActiveSupport::Notifications),
             mapper: mapper,
             subscriptions: subscriptions,
             dispatcher: dispatcher,

rails_event_store/spec/client_spec.rb

@@ -43,5 +43,18 @@ module RailsEventStore
       expect(published.first.metadata[:request_id]).to eq('dummy_id')
       expect(published.first.metadata[:timestamp]).to be_a Time
     end
+
+    specify 'wraps repository into instrumentation' do
+      client = Client.new(repository: InMemoryRepository.new)
+
+      received_notifications = 0
+      ActiveSupport::Notifications.subscribe("append_to_stream.repository.rails_event_store") do
+        received_notifications += 1
+      end
+
+      client.publish(TestEvent.new)
+
+      expect(received_notifications).to eq(1)
+    end
   end
 end

ruby_event_store/lib/ruby_event_store.rb

@@ -24,3 +24,4 @@
 require 'ruby_event_store/async_dispatcher'
 require 'ruby_event_store/debouncer'
 require 'ruby_event_store/version'
+require 'ruby_event_store/instrumented_repository'

ruby_event_store/lib/ruby_event_store/instrumented_repository.rb

@@ -0,0 +1,49 @@
+module RubyEventStore
+  class InstrumentedRepository
+    def initialize(repository, instrumentation)
+      @repository = repository
+      @instrumentation = instrumentation
+    end
+
+    def append_to_stream(events, stream, expected_version)
+      instrumentation.instrument("append_to_stream.repository.rails_event_store", events: events, stream: stream) do
+        repository.append_to_stream(events, stream, expected_version)
+      end
+    end
+
+    def link_to_stream(event_ids, stream, expected_version)
+      instrumentation.instrument("link_to_stream.repository.rails_event_store", event_ids: event_ids, stream: stream) do
+        repository.link_to_stream(event_ids, stream, expected_version)
+      end
+    end
+
+    def delete_stream(stream)
+      instrumentation.instrument("delete_stream.repository.rails_event_store", stream: stream) do
+        repository.delete_stream(stream)
+      end
+    end
+
+    def has_event?(event_id)
+      repository.has_event?(event_id)
+    end
+
+    def last_stream_event(stream)
+      repository.last_stream_event(stream)
+    end
+
+    def read_event(event_id)
+      instrumentation.instrument("read_event.repository.rails_event_store", event_id: event_id) do
+        repository.read_event(event_id)
+      end
+    end
+
+    def read(specification)
+      instrumentation.instrument("read.repository.rails_event_store", specification: specification) do
+        repository.read(specification)
+      end
+    end
+
+    private
+    attr_reader :repository, :instrumentation
+  end
+end

ruby_event_store/ruby_event_store.gemspec

@@ -35,4 +35,5 @@ Gem::Specification.new do |spec|
   spec.add_development_dependency 'unparser'
   spec.add_development_dependency 'astrolabe'
   spec.add_development_dependency 'google-protobuf', '~> 3.5.1.2'
+  spec.add_development_dependency 'activesupport', '~> 5.0'
 end

ruby_event_store/spec/instrumented_repository_spec.rb

@@ -0,0 +1,172 @@
+require 'spec_helper'
+require 'ruby_event_store/spec/event_repository_lint'
+require 'active_support/notifications'
+
+module RubyEventStore
+  RSpec.describe InstrumentedRepository do
+    describe "#append_to_stream" do
+      specify "wraps around original implementation" do
+        some_repository = spy
+        instrumented_repository = InstrumentedRepository.new(some_repository, ActiveSupport::Notifications)
+        event1 = Object.new
+
+        instrumented_repository.append_to_stream([event1], "SomeStream", "c456845d-2b86-49c1-bdef-89e57b5d86b1")
+
+        expect(some_repository).to have_received(:append_to_stream).with([event1], "SomeStream", "c456845d-2b86-49c1-bdef-89e57b5d86b1")
+      end
+
+      specify "instruments" do
+        instrumented_repository = InstrumentedRepository.new(spy, ActiveSupport::Notifications)
+        subscribe_to("append_to_stream.repository.rails_event_store") do |notification_calls|
+          event1 = Object.new
+
+          instrumented_repository.append_to_stream([event1], "SomeStream", "c456845d-2b86-49c1-bdef-89e57b5d86b1")
+
+          expect(notification_calls).to eq([
+            { events: [event1], stream: "SomeStream" }
+          ])
+        end
+      end
+    end
+
+    describe "#link_to_stream" do
+      specify "wraps around original implementation" do
+        some_repository = spy
+        instrumented_repository = InstrumentedRepository.new(some_repository, ActiveSupport::Notifications)
+
+        instrumented_repository.link_to_stream([42], "SomeStream", "c456845d-2b86-49c1-bdef-89e57b5d86b1")
+
+        expect(some_repository).to have_received(:link_to_stream).with([42], "SomeStream", "c456845d-2b86-49c1-bdef-89e57b5d86b1")
+      end
+
+      specify "instruments" do
+        instrumented_repository = InstrumentedRepository.new(spy, ActiveSupport::Notifications)
+        subscribe_to("link_to_stream.repository.rails_event_store") do |notification_calls|
+
+          instrumented_repository.link_to_stream([42], "SomeStream", "c456845d-2b86-49c1-bdef-89e57b5d86b1")
+
+          expect(notification_calls).to eq([
+            { event_ids: [42], stream: "SomeStream" }
+          ])
+        end
+      end
+    end
+
+    describe "#delete_stream" do
+      specify "wraps around original implementation" do
+        some_repository = spy
+        instrumented_repository = InstrumentedRepository.new(some_repository, ActiveSupport::Notifications)
+
+        instrumented_repository.delete_stream("SomeStream")
+
+        expect(some_repository).to have_received(:delete_stream).with("SomeStream")
+      end
+
+      specify "instruments" do
+        instrumented_repository = InstrumentedRepository.new(spy, ActiveSupport::Notifications)
+        subscribe_to("delete_stream.repository.rails_event_store") do |notification_calls|
+
+          instrumented_repository.delete_stream("SomeStream")
+
+          expect(notification_calls).to eq([
+            { stream: "SomeStream" }
+          ])
+        end
+      end
+    end
+
+    describe "#has_event?" do
+      specify "wraps around original implementation" do
+        some_repository = spy
+        instrumented_repository = InstrumentedRepository.new(some_repository, ActiveSupport::Notifications)
+
+        instrumented_repository.has_event?(42)
+
+        expect(some_repository).to have_received(:has_event?).with(42)
+      end
+    end
+
+    describe "#last_stream_event" do
+      specify "wraps around original implementation" do
+        some_repository = spy
+        instrumented_repository = InstrumentedRepository.new(some_repository, ActiveSupport::Notifications)
+
+        instrumented_repository.last_stream_event("SomeStream")
+
+        expect(some_repository).to have_received(:last_stream_event).with("SomeStream")
+      end
+    end
+
+    describe "#read_event" do
+      specify "wraps around original implementation" do
+        some_repository = spy
+        instrumented_repository = InstrumentedRepository.new(some_repository, ActiveSupport::Notifications)
+
+        instrumented_repository.read_event(42)
+
+        expect(some_repository).to have_received(:read_event).with(42)
+      end
+
+      specify "instruments" do
+        instrumented_repository = InstrumentedRepository.new(spy, ActiveSupport::Notifications)
+        subscribe_to("read_event.repository.rails_event_store") do |notification_calls|
+
+          instrumented_repository.read_event(42)
+
+          expect(notification_calls).to eq([
+            { event_id: 42 }
+          ])
+        end
+      end
+    end
+
+    describe "#read" do
+      specify "wraps around original implementation" do
+        some_repository = spy
+        instrumented_repository = InstrumentedRepository.new(some_repository, ActiveSupport::Notifications)
+        specification = double
+
+        instrumented_repository.read(specification)
+
+        expect(some_repository).to have_received(:read).with(specification)
+      end
+
+      specify "instruments" do
+        instrumented_repository = InstrumentedRepository.new(spy, ActiveSupport::Notifications)
+        subscribe_to("read.repository.rails_event_store") do |notification_calls|
+          specification = double
+
+          instrumented_repository.read(specification)
+
+          expect(notification_calls).to eq([
+            { specification: specification }
+          ])
+        end
+      end
+    end
+
+    def subscribe_to(name)
+      received_payloads = []
+      callback = ->(_name, _start, _finish, _id, payload) { received_payloads << payload }
+      ActiveSupport::Notifications.subscribed(callback, name) do
+        yield received_payloads
+      end
+    end
+  end
+end
+
+module RubyEventStore
+  RSpec.describe InstrumentedRepository do
+    subject do
+      InstrumentedRepository.new(InMemoryRepository.new, ActiveSupport::Notifications)
+    end
+
+    let(:test_race_conditions_any)   { false }
+    let(:test_race_conditions_auto)  { false }
+    let(:test_expected_version_auto) { true }
+    let(:test_link_events_to_stream) { true }
+    let(:test_binary) { false }
+
+    it_behaves_like :event_repository, InstrumentedRepository
+  end
+end