lib/good_job/scheduler.rb in good_job-1.8.0 vs lib/good_job/scheduler.rb in good_job-1.9.0

@@ -28,11 +28,11 @@
     }.freeze
 
     # @!attribute [r] instances
     #   @!scope class
     #   List of all instantiated Schedulers in the current process.
-    #   @return [Array<GoodJob:Scheduler>]
+    #   @return [Array<GoodJob::Scheduler>]
     cattr_reader :instances, default: [], instance_reader: false
 
     # Creates GoodJob::Scheduler(s) and Performers from a GoodJob::Configuration instance.
     # @param configuration [GoodJob::Configuration]
     # @param warm_cache_on_initialize [Boolean]
@@ -185,10 +185,12 @@
         active_cache: cache_count,
         available_cache: remaining_cache_count,
       }
     end
 
+    # Preload existing runnable and future-scheduled jobs
+    # @return [void]
     def warm_cache
       return if @max_cache.zero?
 
       performer.next_at(
         limit: @max_cache,
@@ -202,11 +204,11 @@
 
     attr_reader :performer, :executor, :timer_set
 
     def create_executor
       instrument("scheduler_create_pool", { performer_name: performer.name, max_threads: @executor_options[:max_threads] }) do
-        @timer_set = Concurrent::TimerSet.new
+        @timer_set = TimerSet.new
         @executor = ThreadPoolExecutor.new(@executor_options)
       end
     end
 
     def create_task(delay = 0)
@@ -228,11 +230,11 @@
 
       ActiveSupport::Notifications.instrument("#{name}.good_job", payload, &block)
     end
 
     def cache_count
-      timer_set.instance_variable_get(:@queue).length
+      timer_set.length
     end
 
     def remaining_cache_count
       @max_cache - cache_count
     end
@@ -251,9 +253,25 @@
             workers_still_to_be_created = @max_length - @pool.length
             workers_created_but_waiting = @ready.length
             workers_still_to_be_created + workers_created_but_waiting
           end
         end
+      end
+    end
+
+    # Custom sub-class of +Concurrent::TimerSet+ for additional behavior.
+    # @private
+    class TimerSet < Concurrent::TimerSet
+      # Number of scheduled jobs in the queue
+      # @return [Integer]
+      def length
+        @queue.length
+      end
+
+      # Clear the queue
+      # @return [void]
+      def reset
+        synchronize { @queue.clear }
       end
     end
   end
 end