Allow TimerTask to be initialized with a specified Executor

The resulting ScheduledTasks will be initialized with that executor.

Author: bensheldon

lib/concurrent-ruby/concurrent/timer_task.rb

@@ -164,6 +164,7 @@ class TimerTask < RubyExecutorService
     #   @option opts [Boolean] :run_now Whether to run the task immediately
     #     upon instantiation or to wait until the first #  execution_interval
     #     has passed (default: false)
+    #   @option opts [Executor] executor, default is `global_io_executor`
     #
     #   @!macro deref_options
     #
@@ -268,7 +269,8 @@ def ns_initialize(opts, &task)
         warn 'TimeTask timeouts are now ignored as these were not able to be implemented correctly'
       end
       @run_now = opts[:now] || opts[:run_now]
-      @executor = Concurrent::SafeTaskExecutor.new(task)
+      @task = Concurrent::SafeTaskExecutor.new(task)
+      @executor = opts[:executor] || Concurrent.global_io_executor
       @running = Concurrent::AtomicBoolean.new(false)
       @value = nil
 
@@ -289,14 +291,14 @@ def ns_kill_execution
 
     # @!visibility private
     def schedule_next_task(interval = execution_interval)
-      ScheduledTask.execute(interval, args: [Concurrent::Event.new], &method(:execute_task))
+      ScheduledTask.execute(interval, executor: @executor, args: [Concurrent::Event.new], &method(:execute_task))
       nil
     end
 
     # @!visibility private
     def execute_task(completion)
       return nil unless @running.true?
-      _success, value, reason = @executor.execute(self)
+      _success, value, reason = @task.execute(self)
       if completion.try?
         self.value = value
         schedule_next_task

spec/concurrent/timer_task_spec.rb

@@ -82,6 +82,7 @@ def trigger_observable(observable)
           subject = TimerTask.new(execution_interval: 5) { nil }
           expect(subject.execution_interval).to eq 5
         end
+
       end
 
       context '#kill' do
@@ -181,6 +182,33 @@ def trigger_observable(observable)
         expect(latch.count).to eq(0)
         subject.kill
       end
+
+      it 'uses the global executor by default' do
+        executor = Concurrent::ImmediateExecutor.new
+        allow(Concurrent).to receive(:global_io_executor).and_return(executor)
+        allow(executor).to receive(:post).and_call_original
+
+        latch = CountDownLatch.new(1)
+        subject = TimerTask.new(execution_interval: 0.1, run_now: true) { latch.count_down }
+        subject.execute
+        expect(latch.wait(1)).to be_truthy
+        subject.kill
+
+        expect(executor).to have_received(:post)
+      end
+
+      it 'uses a custom executor when given' do
+        executor = Concurrent::ImmediateExecutor.new
+        allow(executor).to receive(:post).and_call_original
+
+        latch = CountDownLatch.new(1)
+        subject = TimerTask.new(execution_interval: 0.1, run_now: true, executor: executor) { latch.count_down }
+        subject.execute
+        expect(latch.wait(1)).to be_truthy
+        subject.kill
+
+        expect(executor).to have_received(:post)
+      end
     end
 
     context 'observation' do