Infer interruption adapter from queue adapter

.rubocop.yml

@@ -12,9 +12,6 @@ AllCops:
 Lint/SuppressedException:
   Exclude:
     - lib/job-iteration.rb
-Style/GlobalVars:
-  Exclude:
-    - lib/job-iteration/integrations/resque.rb
 Naming/FileName:
   Exclude:
     - lib/job-iteration.rb

CHANGELOG.md

@@ -5,6 +5,7 @@
 - [437](https://github.com/Shopify/job-iteration/pull/437) - Use minimum between per-class `job_iteration_max_job_runtime` and `JobIteration.max_job_runtime`, instead of enforcing only setting decreasing values.
   Because it is possible to change the global or parent values after setting the value on a class, it is not possible to truly enforce the decreasing value constraint. Instead, we now use the minimum between the global value and per-class value. This is considered a non-breaking change, as it should not break any **existing** code, it only removes the constraint on new classes.
 - [443](https://github.com/Shopify/job-iteration/pull/443) - Use Sidekiq `:quit` callback to detect graceful shutdown. This makes job-iteration compatible with Sidekiq run in embedded mode.
+- [450](https://github.com/Shopify/job-iteration/pull/450) - Infer which interruption adapter to use from the queue adapter of the job. This deprecates setting `JobIteration.interruption_adapter = <callable>`, in favor of `JobIteration.register_interruption_adapter(<queue adapter name>, <callable>)`. `JobIteration.interruption_adapter` will be removed in a future release.
 
 ### Bug fixes
 

lib/job-iteration.rb

@@ -3,15 +3,12 @@
 require "active_job"
 require_relative "./job-iteration/version"
 require_relative "./job-iteration/enumerator_builder"
+require_relative "./job-iteration/interruption_adapters"
 require_relative "./job-iteration/iteration"
 require_relative "./job-iteration/log_subscriber"
 require_relative "./job-iteration/railtie"
 
 module JobIteration
-  IntegrationLoadError = Class.new(StandardError)
-
-  INTEGRATIONS = [:resque, :sidekiq]
-
   Deprecation = ActiveSupport::Deprecation.new("2.0", "JobIteration")
 
   extend self
@@ -60,10 +57,21 @@ def logger
   # where the throttle backoff value will take precedence over this setting.
   attr_accessor :default_retry_backoff
 
-  # Used internally for hooking into job processing frameworks like Sidekiq and Resque.
-  attr_accessor :interruption_adapter
+  attr_reader :interruption_adapter
 
-  self.interruption_adapter = -> { false }
+  # Overrides interruption checks based on queue adapter.
+  # @deprecated - Use JobIteration::InterruptionAdapters.register(:foo, callable) instead.
+  def interruption_adapter=(adapter)
+    Deprecation.warn("Setting JobIteration.interruption_adapter is deprecated. "\
+      "Use JobIteration::InterruptionAdapters.register(:foo, callable) instead "\
+      "to register the callable (a proc, method, or other object responding to #call) "\
+      "as the interruption adapter for queue adapter :foo.")
+    @interruption_adapter = adapter
+  end
+
+  def register_interruption_adapter(adapter_name, adapter)
+    InterruptionAdapters.register(adapter_name, adapter)
+  end
 
   # Set if you want to use your own enumerator builder instead of default EnumeratorBuilder.
   # @example
@@ -76,29 +84,4 @@ def logger
   attr_accessor :enumerator_builder
 
   self.enumerator_builder = JobIteration::EnumeratorBuilder
-
-  def load_integrations
-    loaded = nil
-    INTEGRATIONS.each do |integration|
-      load_integration(integration)
-      if loaded
-        raise IntegrationLoadError,
-          "#{loaded} integration has already been loaded, but #{integration} is also available. " \
-            "Iteration will only work with one integration."
-      end
-      loaded = integration
-    rescue LoadError
-    end
-  end
-
-  def load_integration(integration)
-    unless INTEGRATIONS.include?(integration)
-      raise IntegrationLoadError,
-        "#{integration} integration is not supported. Available integrations: #{INTEGRATIONS.join(", ")}"
-    end
-
-    require_relative "./job-iteration/integrations/#{integration}"
-  end
 end
-
-JobIteration.load_integrations unless ENV["ITERATION_DISABLE_AUTOCONFIGURE"]

lib/job-iteration/interruption_adapters.rb

@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+require_relative "interruption_adapters/null_adapter"
+
+module JobIteration
+  module InterruptionAdapters
+    BUNDLED_ADAPTERS = [:resque, :sidekiq].freeze # @api private
+
+    class << self
+      # Returns adapter for specified name.
+      #
+      #   JobIteration::InterruptionAdapters.lookup(:sidekiq)
+      #   # => JobIteration::InterruptionAdapters::SidekiqAdapter
+      def lookup(name)
+        registry.fetch(name.to_sym) do
+          Deprecation.warn(<<~DEPRECATION_MESSAGE, caller_locations(1))
+            No interruption adapter is registered for #{name.inspect}; falling back to `NullAdapter`, which never interrupts.
+            Use `JobIteration::InterruptionAdapters.register(#{name.to_sym.inspect}, <adapter>) to register one.
+            This will raise starting in version #{Deprecation.deprecation_horizon} of #{Deprecation.gem_name}!"
+          DEPRECATION_MESSAGE
+
+          NullAdapter
+        end
+      end
+
+      # Registers adapter for specified name.
+      #
+      #   JobIteration::InterruptionAdapters.register(:sidekiq, MyCustomSidekiqAdapter)
+      def register(name, adapter)
+        raise ArgumentError, "adapter must be callable" unless adapter.respond_to?(:call)
+
+        registry[name.to_sym] = adapter
+      end
+
+      private
+
+      attr_reader :registry
+    end
+
+    @registry = {}
+
+    # Built-in Rails adapters. It doesn't make sense to interrupt for these.
+    register(:async, NullAdapter)
+    register(:inline, NullAdapter)
+    register(:test, NullAdapter)
+
+    # External adapters
+    BUNDLED_ADAPTERS.each do |name|
+      require_relative "interruption_adapters/#{name}_adapter"
+    end
+  end
+end

lib/job-iteration/interruption_adapters/null_adapter.rb

@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+# This adapter never interrupts.
+module JobIteration
+  module InterruptionAdapters
+    module NullAdapter
+      class << self
+        def call
+          false
+        end
+      end
+    end
+  end
+end

lib/job-iteration/interruption_adapters/resque_adapter.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+begin
+  require "resque"
+rescue LoadError
+  # Resque is not available, no need to load the adapter
+  return
+end
+
+module JobIteration
+  module InterruptionAdapters
+    module ResqueAdapter
+      # @private
+      module IterationExtension
+        def initialize(*)
+          $resque_worker = self # rubocop:disable Style/GlobalVars
+          super
+        end
+      end
+
+      # @private
+      module ::Resque
+        class Worker
+          # The patch is required in order to call shutdown? on a Resque::Worker instance
+          prepend(IterationExtension)
+        end
+      end
+
+      class << self
+        def call
+          $resque_worker.try!(:shutdown?) # rubocop:disable Style/GlobalVars
+        end
+      end
+    end
+
+    register(:resque, ResqueAdapter)
+  end
+end

lib/job-iteration/interruption_adapters/sidekiq_adapter.rb

@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+begin
+  require "sidekiq"
+rescue LoadError
+  # Sidekiq is not available, no need to load the adapter
+  return
+end
+
+module JobIteration
+  module InterruptionAdapters
+    module SidekiqAdapter
+      class << self
+        attr_accessor :stopping
+
+        def call
+          stopping
+        end
+      end
+
+      ::Sidekiq.configure_server do |config|
+        config.on(:quiet) do
+          SidekiqAdapter.stopping = true
+        end
+      end
+    end
+
+    register(:sidekiq, SidekiqAdapter)
+  end
+end

lib/job-iteration/iteration.rb

@@ -120,6 +120,10 @@ def retry_job(*, **)
 
     private
 
+    def interruption_adapter # @private
+      JobIteration.interruption_adapter || JobIteration::InterruptionAdapters.lookup(self.class.queue_adapter_name)
+    end
+
     def enumerator_builder
       JobIteration.enumerator_builder.new(self)
     end
@@ -280,7 +284,7 @@ def job_should_exit?
       max_job_runtime = job_iteration_max_job_runtime
       return true if max_job_runtime && start_time && (Time.now.utc - start_time) > max_job_runtime
 
-      JobIteration.interruption_adapter.call || (defined?(super) && super)
+      interruption_adapter.call || (defined?(super) && super)
     end
 
     def job_iteration_max_job_runtime

test/integration/integration_behaviour.rb

@@ -55,8 +55,8 @@ module IntegrationBehaviour
   private
 
   # Should return the symbol to use when configuring the adapter
-  #     ActiveJob::Base.queue_adapter = adapter
-  def adapter
+  #     ActiveJob::Base.queue_adapter = queue_adapter
+  def queue_adapter
     raise NotImplemented, "#{self.class.name} must implement #{__method__}"
   end