ScheduledTask noe uses global timer pool.

Author: jdantonio

lib/concurrent/executor/timer_set.rb

@@ -70,7 +70,7 @@ def post(intended_time, &block)
       @mutex.synchronize do
         return false if shutdown?
         raise ArgumentError.new('no block given') unless block_given?
-        time = calculate_schedule_time(intended_time)
+        time = TimerSet.calculate_schedule_time(intended_time).to_f
 
         if (time - Time.now.to_f) <= 0.01
           @executor.post(&block)
@@ -82,6 +82,9 @@ def post(intended_time, &block)
       true
     end
 
+    # Begin an orderly shutdown. Tasks already in the queue will be executed,
+    # but no new tasks will be accepted. Has no additional effect if the
+    # thread pool is not running.
     def shutdown
       @mutex.synchronize do
         unless @shutdown.set?
@@ -94,6 +97,29 @@ def shutdown
     end
     alias_method :kill, :shutdown
 
+    # Calculate an Epoch time with milliseconds at which to execute a
+    # task. If the given time is a `Time` object it will be converted
+    # accordingly. If the time is an integer value greate than zero
+    # it will be understood as a number of seconds in the future and
+    # will be added to the current time to calculate Epoch.
+    #
+    # @param [Object] intended_time the time (as a `Time` object or an integer)
+    #   to schedule the task for execution
+    # @param [Time] now (Time.now) the time from which to calculate an interval
+    #
+    # @return [Fixnum] the intended time as seconds/millis from Epoch
+    #
+    # @raise [ArgumentError] if the intended execution time is not in the future
+    def self.calculate_schedule_time(intended_time, now = Time.now)
+      if intended_time.is_a?(Time)
+        raise ArgumentError.new('schedule time must be in the future') if intended_time <= now
+        intended_time
+      else
+        raise ArgumentError.new('seconds must be greater than zero') if intended_time.to_f < 0.0
+        now + intended_time
+      end
+    end
+
     private
 
     # A struct for encapsulating a task and its intended execution time.
@@ -109,25 +135,6 @@ def <=>(other)
       end
     end
 
-    # Calculate an Epoch time with milliseconds at which to execute a
-    # task. If the given time is a `Time` object it will be converted
-    # accordingly. If the time is an integer value greate than zero
-    # it will be understood as a number of seconds in the future and
-    # will be added to the current time to calculate Epoch.
-    #
-    # @raise [ArgumentError] if the intended execution time is not in the future
-    #
-    # @!visibility private
-    def calculate_schedule_time(intended_time, now = Time.now)
-      if intended_time.is_a?(Time)
-        raise ArgumentError.new('schedule time must be in the future') if intended_time <= now
-        intended_time.to_f
-      else
-        raise ArgumentError.new('seconds must be greater than zero') if intended_time.to_f < 0.0
-        now.to_f + intended_time.to_f
-      end
-    end
-
     # Check the status of the processing thread. This thread is responsible
     # for monitoring the internal task queue and sending tasks to the
     # executor when it is time for them to be processed. If there is no

lib/concurrent/scheduled_task.rb

@@ -6,34 +6,31 @@ module Concurrent
 
   class ScheduledTask < IVar
 
-    SchedulingError = Class.new(ArgumentError)
-
     attr_reader :schedule_time
 
-    def initialize(schedule_time, opts = {}, &block)
-      raise SchedulingError.new('no block given') unless block_given?
-      calculate_schedule_time!(schedule_time) # raise exception if in past
+    def initialize(intended_time, opts = {}, &block)
+      raise ArgumentError.new('no block given') unless block_given?
+      TimerSet.calculate_schedule_time(intended_time) # raises exceptons
 
       super(NO_VALUE, opts)
 
+      @intended_time =  intended_time
       @state = :unscheduled
-      @intended_schedule_time = schedule_time
-      @schedule_time = nil
       @task = block
     end
 
     # @since 0.5.0
     def execute
       if compare_and_set_state(:pending, :unscheduled)
-        @schedule_time = calculate_schedule_time!(@intended_schedule_time).freeze
-        do_next_interval
+        @schedule_time = TimerSet.calculate_schedule_time(@intended_time)
+        Concurrent::timer(@schedule_time.to_f - Time.now.to_f, &method(:process_task))
         self
       end
     end
 
     # @since 0.5.0
-    def self.execute(schedule_time, opts = {}, &block)
-      return ScheduledTask.new(schedule_time, opts, &block).execute
+    def self.execute(intended_time, opts = {}, &block)
+      return ScheduledTask.new(intended_time, opts, &block).execute
     end
 
     def cancelled?
@@ -63,7 +60,7 @@ def add_observer(*args)
 
     private
 
-    def do_work
+    def process_task
       if compare_and_set_state(:in_progress, :pending)
         success, val, reason = SafeTaskExecutor.new(@task).execute
 
@@ -76,29 +73,5 @@ def do_work
         observers.notify_and_delete_observers{ [time, self.value, reason] }
       end
     end
-
-    def do_next_interval
-      return if cancelled?
-
-      interval = mutex.synchronize do
-        [60, [(@schedule_time.to_f - Time.now.to_f), 0].max].min
-      end
-
-      if interval > 0
-        Concurrent::timer(interval, &method(:do_next_interval))
-      else
-        do_work
-      end
-    end
-
-    def calculate_schedule_time!(schedule_time, now = Time.now)
-      if schedule_time.is_a?(Time)
-        raise SchedulingError.new('schedule time must be in the future') if schedule_time <= now
-        schedule_time.dup
-      else
-        raise SchedulingError.new('seconds must be greater than zero') if schedule_time.to_f <= 0.0
-        now + schedule_time.to_f
-      end
-    end
   end
 end

lib/concurrent/utilities/timer.rb

@@ -6,7 +6,9 @@ module Concurrent
   # Perform the given operation asynchronously after the given number of seconds.
   #
   # @param [Fixnum] seconds the interval in seconds to wait before executing the task
+  #
   # @yield the task to execute
+  #
   # @return [Boolean] true
   def timer(seconds, &block)
     raise ArgumentError.new('no block given') unless block_given?

spec/concurrent/scheduled_task_spec.rb

@@ -65,19 +65,19 @@ def execute_dereferenceable(subject)
       it 'raises an exception when seconds is less than zero' do
         expect {
           ScheduledTask.new(-1){ nil }
-        }.to raise_error(ScheduledTask::SchedulingError)
+        }.to raise_error(ArgumentError)
       end
 
       it 'raises an exception when schedule time is in the past' do
         expect {
           ScheduledTask.new(Time.now - 60){ nil }
-        }.to raise_error(ScheduledTask::SchedulingError)
+        }.to raise_error(ArgumentError)
       end
 
       it 'raises an exception when no block given' do
         expect {
           ScheduledTask.new(1)
-        }.to raise_error(ScheduledTask::SchedulingError)
+        }.to raise_error(ArgumentError)
       end
 
       it 'sets the initial state to :unscheduled' do
@@ -121,7 +121,7 @@ def execute_dereferenceable(subject)
           Timecop.travel(60)
           expect {
             task.execute
-          }.to raise_error(ScheduledTask::SchedulingError)
+          }.to raise_error(ArgumentError)
         end
       end