Parameterize clock when initializing repository

Instead of passing it every time.

Author: swistak35

Diff for: contrib/ruby_event_store-outbox/lib/ruby_event_store/outbox/consumer.rb

@@ -24,7 +24,7 @@ def initialize(consumer_uuid, configuration, clock: Time, logger:, metrics:)
         raise "Unknown format" if configuration.message_format != SIDEKIQ5_FORMAT
         @processor = SidekiqProcessor.new(Redis.new(url: configuration.redis_url))
 
-        @repository = Repositories::Mysql57.build_for_consumer(configuration.database_url)
+        @repository = Repositories::Mysql57.build_for_consumer(configuration.database_url, clock: clock)
         @cleanup_strategy = CleanupStrategies.build(configuration, repository)
       end
 
@@ -123,7 +123,7 @@ def log_error(e)
       end
 
       def obtain_lock_for_process(fetch_specification)
-        result = repository.obtain_lock_for_process(fetch_specification, consumer_uuid, clock: @clock)
+        result = repository.obtain_lock_for_process(fetch_specification, consumer_uuid)
         case result
         when :deadlocked
           logger.warn "Obtaining lock for split_key '#{fetch_specification.split_key}' failed (deadlock)"
@@ -158,7 +158,7 @@ def release_lock_for_process(fetch_specification)
       end
 
       def refresh_lock_for_process(lock)
-        result = repository.refresh_lock(lock, clock: @clock)
+        result = repository.refresh_lock(lock)
         case result
         when :ok
           return true

Diff for: contrib/ruby_event_store-outbox/lib/ruby_event_store/outbox/repositories/mysql57.rb

@@ -115,12 +115,20 @@ def self.get_lock_record(fetch_specification)
           end
         end
 
-        def self.build_for_consumer(database_url)
+        def self.build_for_consumer(database_url, clock:)
           ::ActiveRecord::Base.establish_connection(database_url) unless ::ActiveRecord::Base.connected?
           if ::ActiveRecord::Base.connection.adapter_name == "Mysql2"
             ::ActiveRecord::Base.connection.execute("SET SESSION innodb_lock_wait_timeout = 1;")
           end
-          new
+          new(clock: clock)
+        end
+
+        def self.build_for_producer(clock: Time)
+          new(clock: clock)
+        end
+
+        def initialize(clock:)
+          @clock = clock
         end
 
         def insert_record(format, split_key, payload)
@@ -135,15 +143,15 @@ def get_remaining_count(fetch_specification)
           Record.remaining_for(fetch_specification).count
         end
 
-        def obtain_lock_for_process(fetch_specification, process_uuid, clock:)
+        def obtain_lock_for_process(fetch_specification, process_uuid)
           Lock.obtain(fetch_specification, process_uuid, clock: clock)
         end
 
         def release_lock_for_process(fetch_specification, process_uuid)
           Lock.release(fetch_specification, process_uuid)
         end
 
-        def refresh_lock(lock, clock:)
+        def refresh_lock(lock)
           Lock.refresh(lock, clock: clock)
         end
 
@@ -161,6 +169,9 @@ def delete_enqueued_older_than(fetch_specification, duration, limit)
         rescue ::ActiveRecord::LockWaitTimeout
           :lock_timeout
         end
+
+        private
+        attr_reader :clock
       end
     end
   end

Diff for: contrib/ruby_event_store-outbox/lib/ruby_event_store/outbox/sidekiq_producer.rb

@@ -7,7 +7,7 @@
 module RubyEventStore
   module Outbox
     class SidekiqProducer
-      def initialize(repository = Repositories::Mysql57.new)
+      def initialize(repository)
         @repository = repository
       end
 

Diff for: contrib/ruby_event_store-outbox/lib/ruby_event_store/outbox/sidekiq_scheduler.rb

@@ -5,7 +5,7 @@
 module RubyEventStore
   module Outbox
     class SidekiqScheduler
-      def initialize(repository: Repositories::Mysql57.new, serializer: RubyEventStore::Serializers::YAML)
+      def initialize(repository: Repositories::Mysql57.build_for_producer, serializer: RubyEventStore::Serializers::YAML)
         @serializer = serializer
         @sidekiq_producer = SidekiqProducer.new(repository)
       end

Diff for: contrib/ruby_event_store-outbox/spec/consumer_spec.rb

@@ -36,8 +36,8 @@ module Outbox
       end
 
       specify "push the jobs to sidekiq" do
-        record = create_record("default", "default")
         clock = TickingClock.new
+        record = create_record("default", "default", clock: clock)
         consumer =
           Consumer.new(SecureRandom.uuid, default_configuration, clock: clock, logger: logger, metrics: null_metrics)
         result = consumer.process
@@ -53,10 +53,10 @@ module Outbox
       end
 
       specify "push multiple jobs to different queues" do
-        create_record("default", "default")
-        create_record("default", "default")
-        create_record("default2", "default2")
         clock = TickingClock.new
+        create_record("default", "default", clock: clock)
+        create_record("default", "default", clock: clock)
+        create_record("default2", "default2", clock: clock)
         consumer =
           Consumer.new(SecureRandom.uuid, default_configuration, clock: clock, logger: logger, metrics: null_metrics)
 
@@ -87,9 +87,9 @@ module Outbox
       end
 
       specify "returns false if didnt aquire lock" do
-        create_record("default", "default")
-        consumer = Consumer.new(SecureRandom.uuid, default_configuration, logger: logger, metrics: null_metrics)
         clock = TickingClock.new
+        create_record("default", "default", clock: clock)
+        consumer = Consumer.new(SecureRandom.uuid, default_configuration, logger: logger, metrics: null_metrics)
         Repositories::Mysql57::Lock.obtain(
           FetchSpecification.new(SIDEKIQ5_FORMAT, "default"),
           "some-other-process-uuid",
@@ -168,10 +168,10 @@ module Outbox
       end
 
       specify "incorrect payload wont cause later messages to schedule" do
-        record1 = create_record("default", "default")
-        record1.update!(payload: "unparsable garbage")
-        record2 = create_record("default", "default")
         clock = TickingClock.new
+        record1 = create_record("default", "default", clock: clock)
+        record1.update!(payload: "unparsable garbage")
+        record2 = create_record("default", "default", clock: clock)
         consumer =
           Consumer.new(SecureRandom.uuid, default_configuration, clock: clock, logger: logger, metrics: null_metrics)
 
@@ -245,15 +245,15 @@ module Outbox
       end
 
       specify "deadlock when releasing lock doesnt do anything" do
-        create_record("default", "default")
+        clock = TickingClock.new
+        create_record("default", "default", clock: clock)
         allow(Repositories::Mysql57::Lock).to receive(:lock).and_wrap_original do |m, *args|
           if caller.any? { |l| l.include? "`release'" }
             raise ::ActiveRecord::Deadlocked
           else
             m.call(*args)
           end
         end
-        clock = TickingClock.new
         consumer =
           Consumer.new(
             SecureRandom.uuid,
@@ -272,7 +272,6 @@ module Outbox
       end
 
       specify "lock timeout when releasing lock doesnt do anything" do
-        create_record("default", "default")
         allow(Repositories::Mysql57::Lock).to receive(:lock).and_wrap_original do |m, *args|
           if caller.any? { |l| l.include? "`release'" }
             raise ::ActiveRecord::LockWaitTimeout
@@ -281,6 +280,7 @@ module Outbox
           end
         end
         clock = TickingClock.new
+        create_record("default", "default", clock: clock)
         consumer =
           Consumer.new(
             SecureRandom.uuid,
@@ -299,8 +299,8 @@ module Outbox
       end
 
       specify "after successful loop, lock is released" do
-        create_record("default", "default")
         clock = TickingClock.new
+        create_record("default", "default", clock: clock)
         consumer =
           Consumer.new(SecureRandom.uuid, default_configuration, clock: clock, logger: logger, metrics: null_metrics)
 
@@ -312,8 +312,8 @@ module Outbox
       end
 
       specify "lock disappearing in the meantime, doesnt do anything" do
-        create_record("default", "default")
         clock = TickingClock.new
+        create_record("default", "default", clock: clock)
         consumer =
           Consumer.new(SecureRandom.uuid, default_configuration, clock: clock, logger: logger, metrics: test_metrics)
         allow(consumer).to receive(:release_lock_for_process).and_wrap_original do |m, *args|
@@ -332,8 +332,8 @@ module Outbox
       end
 
       specify "lock stolen in the meantime, doesnt do anything" do
-        create_record("default", "default")
         clock = TickingClock.new
+        create_record("default", "default", clock: clock)
         consumer =
           Consumer.new(SecureRandom.uuid, default_configuration, clock: clock, logger: logger, metrics: null_metrics)
         allow(consumer).to receive(:release_lock_for_process).and_wrap_original do |m, *args|
@@ -349,12 +349,13 @@ module Outbox
       end
 
       specify "old lock can be reobtained" do
+        clock = TickingClock.new(start: 10.minutes.ago)
         Repositories::Mysql57::Lock.obtain(
           FetchSpecification.new(SIDEKIQ5_FORMAT, "default"),
           "some-old-uuid",
-          clock: TickingClock.new(start: 10.minutes.ago)
+          clock: clock
         )
-        record = create_record("default", "default")
+        record = create_record("default", "default", clock: clock)
         consumer = Consumer.new(SecureRandom.uuid, default_configuration, logger: logger, metrics: null_metrics)
 
         result = consumer.process
@@ -365,10 +366,11 @@ module Outbox
       end
 
       specify "relatively fresh locks are not reobtained" do
+        clock = TickingClock.new(start: 9.minutes.ago)
         Repositories::Mysql57::Lock.obtain(
           FetchSpecification.new(SIDEKIQ5_FORMAT, "default"),
           "some-old-uuid",
-          clock: TickingClock.new(start: 9.minutes.ago)
+          clock: clock
         )
         create_record("default", "default")
         consumer = Consumer.new(SecureRandom.uuid, default_configuration, logger: logger, metrics: null_metrics)
@@ -379,8 +381,8 @@ module Outbox
       end
 
       specify "when inserting lock, other process may do same concurrently" do
-        record = create_record("default", "default")
         clock = TickingClock.new
+        record = create_record("default", "default", clock: clock)
         consumer =
           Consumer.new(SecureRandom.uuid, default_configuration, clock: clock, logger: logger, metrics: null_metrics)
         allow(Repositories::Mysql57::Lock).to receive(:create!).and_wrap_original do |m, *args|
@@ -464,8 +466,8 @@ module Outbox
       end
 
       specify "clean old jobs" do
-        create_record("default", "default")
         clock = TickingClock.new
+        create_record("default", "default", clock: clock)
         consumer =
           Consumer.new(
             SecureRandom.uuid,
@@ -485,8 +487,8 @@ module Outbox
       end
 
       specify "clean old jobs with limit" do
-        3.times.map { create_record("default", "default") }
         clock = TickingClock.new
+        3.times.map { create_record("default", "default", clock: clock) }
         consumer =
           Consumer.new(
             SecureRandom.uuid,
@@ -506,8 +508,8 @@ module Outbox
       end
 
       specify "clean old jobs - lock timeout" do
-        create_record("default", "default")
         clock = TickingClock.new
+        create_record("default", "default", clock: clock)
         consumer =
           Consumer.new(
             SecureRandom.uuid,
@@ -530,8 +532,8 @@ module Outbox
       end
 
       specify "clean old jobs - deadlock" do
-        create_record("default", "default")
         clock = TickingClock.new
+        create_record("default", "default", clock: clock)
         consumer =
           Consumer.new(
             SecureRandom.uuid,
@@ -553,7 +555,7 @@ module Outbox
         expect(test_metrics.operation_results).to include({ operation: "cleanup", result: "deadlocked" })
       end
 
-      def create_record(queue, split_key, format: "sidekiq5")
+      def create_record(queue, split_key, format: "sidekiq5", clock: Time)
         payload = {
           class: "SomeAsyncHandler",
           queue: queue,
@@ -569,7 +571,7 @@ def create_record(queue, split_key, format: "sidekiq5")
             }
           ]
         }
-        Repositories::Mysql57.new.insert_record(format, split_key, payload.to_json)
+        Repositories::Mysql57.build_for_producer(clock: clock).insert_record(format, split_key, payload.to_json)
       end
     end
   end