Extract Supervised concern from Runnable

And set the mode explicitly, not in #start

Author: rosa

lib/solid_queue.rb

@@ -5,6 +5,7 @@
 require "active_job/concurrency_controls"
 
 require "solid_queue/app_executor"
+require "solid_queue/processes/supervised"
 require "solid_queue/processes/registrable"
 require "solid_queue/interruptible"
 require "solid_queue/pidfile"

lib/solid_queue/processes/registrable.rb

@@ -10,28 +10,22 @@ module Registrable
 
       set_callback :shutdown, :before, :stop_heartbeat
       set_callback :shutdown, :after, :deregister
-
-      attr_reader :supervisor
     end
 
     def inspect
       "#{kind}(pid=#{process_pid}, hostname=#{hostname}, metadata=#{metadata})"
     end
     alias to_s inspect
 
-    def supervised_by(process)
-      @supervisor = process
-    end
-
     private
       attr_accessor :process
 
       def register
         @process = SolidQueue::Process.register \
           kind: self.class.name.demodulize,
           pid: process_pid,
-          supervisor: supervisor,
           hostname: hostname,
+          supervisor: try(:supervisor),
           metadata: metadata
       end
 

lib/solid_queue/processes/runnable.rb

@@ -1,107 +1,84 @@
 # frozen_string_literal: true
 
-module SolidQueue
-  module Processes
-    module Runnable
-      def start(mode: :supervised)
-        @mode = mode.to_s.inquiry
-        @stopping = false
+module SolidQueue::Processes
+  module Runnable
+    include Supervised
 
-        observe_initial_delay
-        run_callbacks(:boot) { boot }
+    def start
+      @stopping = false
 
-        start_loop
-      end
-
-      def stop
-        @stopping = true
-        @thread&.join
-      end
-
-    private
-      attr_reader :mode
+      observe_initial_delay
+      run_callbacks(:boot) { boot }
 
-      def boot
-        register_signal_handlers
-        SolidQueue.logger.info("[SolidQueue] Starting #{self}")
-      end
-
-      def register_signal_handlers
-        %w[ INT TERM ].each do |signal|
-          trap(signal) do
-            stop
-            interrupt
-          end
-        end
-
-        trap(:QUIT) do
-          exit!
-        end
-      end
+      start_loop
+    end
 
-      def start_loop
-        if mode.async?
-          @thread = Thread.new { do_start_loop }
-        else
-          do_start_loop
-        end
-      end
+    def stop
+      @stopping = true
+      @thread&.join
+    end
 
-      def do_start_loop
-        procline "started"
+  private
+    attr_writer :mode
 
-        loop do
-          break if shutting_down?
+    DEFAULT_MODE = :async
 
-          run
-        end
-      ensure
-        run_callbacks(:shutdown) { shutdown }
-      end
+    def mode
+      (@mode || DEFAULT_MODE).to_s.inquiry
+    end
 
-      def shutting_down?
-        stopping? || supervisor_went_away? || finished?
-      end
+    def boot
+      register_signal_handlers if supervised?
+      SolidQueue.logger.info("[SolidQueue] Starting #{self}")
+    end
 
-      def run
-        raise NotImplementedError
+    def start_loop
+      if mode.async?
+        @thread = Thread.new { do_start_loop }
+      else
+        do_start_loop
       end
+    end
 
-      def shutdown
-        procline "shutting down"
+    def do_start_loop
+      loop do
+        break if shutting_down?
 
+        run
       end
+    ensure
+      run_callbacks(:shutdown) { shutdown }
+    end
 
-      def stopping?
-        @stopping
-      end
+    def shutting_down?
+      stopping? || supervisor_went_away? || finished?
+    end
 
-      def finished?
-        running_inline? && all_work_completed?
-      end
+    def run
+      raise NotImplementedError
+    end
 
-      def supervisor_went_away?
-        supervised? && supervisor&.pid != ::Process.ppid
-      end
+    def stopping?
+      @stopping
+    end
 
-      def supervised?
-        mode.supervised?
-      end
+    def finished?
+      running_inline? && all_work_completed?
+    end
 
-      def all_work_completed?
-        false
-      end
+    def all_work_completed?
+      false
+    end
 
-      def running_inline?
-        mode.inline?
-      end
+    def running_inline?
+      mode.inline?
+    end
 
-      def with_polling_volume
-        if SolidQueue.silence_polling?
-          ActiveRecord::Base.logger.silence { yield }
-        else
-          yield
-        end
+    def with_polling_volume
+      if SolidQueue.silence_polling?
+        ActiveRecord::Base.logger.silence { yield }
+      else
+        yield
       end
     end
   end

lib/solid_queue/processes/supervised.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+module SolidQueue::Processes
+  module Supervised
+    extend ActiveSupport::Concern
+
+    included do
+      attr_reader :supervisor
+    end
+
+    def supervised_by(process)
+      self.mode = :supervised
+      @supervisor = process
+    end
+
+    private
+      def supervisor_went_away?
+        supervised? && supervisor&.pid != ::Process.ppid
+      end
+
+      def supervised?
+        mode.supervised?
+      end
+
+      def register_signal_handlers
+        %w[ INT TERM ].each do |signal|
+          trap(signal) do
+            stop
+            interrupt
+          end
+        end
+
+        trap(:QUIT) do
+          exit!
+        end
+      end
+  end
+end

test/integration/jobs_lifecycle_test.rb

@@ -18,8 +18,8 @@ class JobsLifecycleTest < ActiveSupport::TestCase
     AddToBufferJob.perform_later "hey"
     AddToBufferJob.perform_later "ho"
 
-    @dispatcher.start(mode: :async)
-    @worker.start(mode: :async)
+    @dispatcher.start
+    @worker.start
 
     wait_for_jobs_to_finish_for(2.seconds)
 
@@ -31,8 +31,8 @@ class JobsLifecycleTest < ActiveSupport::TestCase
     AddToBufferJob.set(wait: 1.day).perform_later("I'm scheduled")
     AddToBufferJob.set(wait: 3.days).perform_later("I'm scheduled later")
 
-    @dispatcher.start(mode: :async)
-    @worker.start(mode: :async)
+    @dispatcher.start
+    @worker.start
 
     assert_equal 2, SolidQueue::ScheduledExecution.count
 
@@ -56,7 +56,7 @@ class JobsLifecycleTest < ActiveSupport::TestCase
   test "delete finished jobs after they run" do
     deleting_finished_jobs do
       AddToBufferJob.perform_later "hey"
-      @worker.start(mode: :async)
+      @worker.start
 
       wait_for_jobs_to_finish_for(2.seconds)
     end

test/unit/dispatcher_test.rb

@@ -17,7 +17,7 @@ class DispatcherTest < ActiveSupport::TestCase
     old_logger, ActiveRecord::Base.logger = ActiveRecord::Base.logger, ActiveSupport::Logger.new(log)
     old_silence_polling, SolidQueue.silence_polling = SolidQueue.silence_polling, false
 
-    @dispatcher.start(mode: :async)
+    @dispatcher.start
     sleep 0.5
 
     assert_match /SELECT .* FROM .solid_queue_scheduled_executions. WHERE/, log.string
@@ -31,7 +31,7 @@ class DispatcherTest < ActiveSupport::TestCase
     old_logger, ActiveRecord::Base.logger = ActiveRecord::Base.logger, ActiveSupport::Logger.new(log)
     old_silence_polling, SolidQueue.silence_polling = SolidQueue.silence_polling, true
 
-    @dispatcher.start(mode: :async)
+    @dispatcher.start
     sleep 0.5
 
     assert_no_match /SELECT .* FROM .solid_queue_scheduled_executions. WHERE/, log.string
@@ -47,8 +47,8 @@ class DispatcherTest < ActiveSupport::TestCase
     assert_equal 15, SolidQueue::ScheduledExecution.count
 
     another_dispatcher = SolidQueue::Dispatcher.new(polling_interval: 0.1, batch_size: 10)
-    @dispatcher.start(mode: :async)
-    another_dispatcher.start(mode: :async)
+    @dispatcher.start
+    another_dispatcher.start
 
     sleep 0.5
 

test/unit/worker_test.rb

@@ -23,7 +23,7 @@ class WorkerTest < ActiveSupport::TestCase
 
     AddToBufferJob.perform_later "hey!"
 
-    @worker.start(mode: :async)
+    @worker.start
 
     wait_for_jobs_to_finish_for(1.second)
     @worker.wake_up
@@ -44,7 +44,7 @@ class WorkerTest < ActiveSupport::TestCase
       StoreResultJob.perform_later(:immediate)
     end
 
-    @worker.start(mode: :async)
+    @worker.start
 
     wait_for_jobs_to_finish_for(1.second)
     @worker.wake_up
@@ -58,7 +58,7 @@ class WorkerTest < ActiveSupport::TestCase
     old_logger, ActiveRecord::Base.logger = ActiveRecord::Base.logger, ActiveSupport::Logger.new(log)
     old_silence_polling, SolidQueue.silence_polling = SolidQueue.silence_polling, false
 
-    @worker.start(mode: :async)
+    @worker.start
     sleep 0.2
 
     assert_match /SELECT .* FROM .solid_queue_ready_executions. WHERE .solid_queue_ready_executions...queue_name./, log.string
@@ -72,7 +72,7 @@ class WorkerTest < ActiveSupport::TestCase
     old_logger, ActiveRecord::Base.logger = ActiveRecord::Base.logger, ActiveSupport::Logger.new(log)
     old_silence_polling, SolidQueue.silence_polling = SolidQueue.silence_polling, true
 
-    @worker.start(mode: :async)
+    @worker.start
     sleep 0.2
 
     assert_no_match /SELECT .* FROM .solid_queue_ready_executions. WHERE .solid_queue_ready_executions...queue_name./, log.string