Run non-standalone supervisor in a separate thread

Otherwise `supervise` will block and we won't be able to stop it
externally (from the Puma plugin).

Author: rosa

lib/solid_queue/app_executor.rb

@@ -17,5 +17,15 @@ def handle_thread_error(error)
         SolidQueue.on_thread_error.call(error)
       end
     end
+
+    def create_thread(&block)
+      Thread.new do
+        Thread.current.name = name
+        block.call
+      rescue Exception => exception
+        handle_thread_error(exception)
+        raise
+      end
+    end
   end
 end

lib/solid_queue/async_supervisor.rb

@@ -2,36 +2,47 @@
 
 module SolidQueue
   class AsyncSupervisor < Supervisor
+    def stop
+      super
+      @thread&.join
+    end
+
     private
+      def supervise
+        if standalone? then super
+        else
+          @thread = create_thread { super }
+        end
+      end
 
-    def check_and_replace_terminated_processes
-      terminated_threads = process_instances.select { |thread_id, instance| !instance.alive? }
-      terminated_threads.each { |thread_id, instance| replace_thread(thread_id, instance) }
-    end
+      def check_and_replace_terminated_processes
+        terminated_threads = process_instances.select { |thread_id, instance| !instance.alive? }
+        terminated_threads.each { |thread_id, instance| replace_thread(thread_id, instance) }
+      end
 
-    def replace_thread(thread_id, instance)
-      SolidQueue.instrument(:replace_thread, supervisor_pid: ::Process.pid) do |payload|
-        payload[:thread] = instance
+      def replace_thread(thread_id, instance)
+        SolidQueue.instrument(:replace_thread, supervisor_pid: ::Process.pid) do |payload|
+          payload[:thread] = instance
 
-        error = Processes::ThreadTerminatedError.new(terminated_instance.name)
-        release_claimed_jobs_by(terminated_instance, with_error: error)
+          error = Processes::ThreadTerminatedError.new(terminated_instance.name)
+          release_claimed_jobs_by(terminated_instance, with_error: error)
 
-        start_process(configured_processes.delete(thread_id))
+          start_process(configured_processes.delete(thread_id))
+        end
       end
-    end
 
-    def perform_graceful_termination
-      process_instances.values.each(&:stop)
+      def perform_graceful_termination
+        process_instances.values.each(&:stop)
 
-      Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? })
-    end
+        Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? })
+      end
 
-    def perform_immediate_termination
-      exit!
-    end
+      def perform_immediate_termination
+        exit!
+      end
 
-    def all_processes_terminated?
-      process_instances.values.none?(&:alive?)
-    end
+      def all_processes_terminated?
+        process_instances.values.none?(&:alive?)
+      end
   end
 end

lib/solid_queue/dispatcher.rb

@@ -3,6 +3,7 @@
 module SolidQueue
   class Dispatcher < Processes::Poller
     include LifecycleHooks
+
     attr_reader :batch_size
 
     after_boot :run_start_hooks

lib/solid_queue/processes/runnable.rb

@@ -92,16 +92,5 @@ def running_async?
       def running_as_fork?
         mode.fork?
       end
-
-
-      def create_thread(&block)
-        Thread.new do
-          Thread.current.name = name
-          block.call
-        rescue Exception => exception
-          handle_thread_error(exception)
-          raise
-        end
-      end
   end
 end