wip: Java executors now share JavaExecutor mixin module.

Author: jdantonio

lib/concurrent/executor/executor.rb

@@ -44,16 +44,6 @@ def wait_for_termination(timeout)
     def post(*args, &task)
     end
 
-    # Submit a task to the thread pool for asynchronous processing.
-    #
-    # @param [Proc] task the asynchronous task to perform
-    #
-    # @return [self] returns itself
-    def <<(task)
-      post(&task)
-      self
-    end
-
     protected
 
     attr_reader :mutex, :stop_event, :stopped_event
@@ -67,4 +57,93 @@ def init_executor
     def execute(*args, &task)
     end
   end
+
+  if RUBY_PLATFORM == 'java'
+
+    module JavaExecutor
+
+      # Submit a task to the thread pool for asynchronous processing.
+      #
+      # @param [Array] args zero or more arguments to be passed to the task
+      #
+      # @yield the asynchronous task to perform
+      #
+      # @return [Boolean] `true` if the task is queued, `false` if the thread pool
+      #   is not running
+      #
+      # @raise [ArgumentError] if no task is given
+      def post(*args)
+        raise ArgumentError.new('no block given') unless block_given?
+        if running?
+          @executor.submit{ yield(*args) }
+          true
+        else
+          false
+        end
+      rescue Java::JavaUtilConcurrent::RejectedExecutionException => ex
+        raise RejectedExecutionError
+      end
+
+      # Submit a task to the thread pool for asynchronous processing.
+      #
+      # @param [Proc] task the asynchronous task to perform
+      #
+      # @return [self] returns itself
+      def <<(task)
+        post(&task)
+        self
+      end
+
+      # Is the thread pool running?
+      #
+      # @return [Boolean] `true` when running, `false` when shutting down or shutdown
+      def running?
+        ! (@executor.isShutdown || @executor.isTerminated)
+      end
+
+      # Is the thread pool shutdown?
+      #
+      # @return [Boolean] `true` when shutdown, `false` when shutting down or running
+      def shutdown?
+        @executor.isShutdown
+      end
+
+      # Block until thread pool shutdown is complete or until `timeout` seconds have
+      # passed.
+      #
+      # @note Does not initiate shutdown or termination. Either `shutdown` or `kill`
+      #   must be called before this method (or on another thread).
+      #
+      # @param [Integer] timeout the maximum number of seconds to wait for shutdown to complete
+      #
+      # @return [Boolean] `true` if shutdown complete or false on `timeout`
+      def wait_for_termination(timeout)
+        @executor.awaitTermination(timeout.to_i, java.util.concurrent.TimeUnit::SECONDS)
+      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
+        @executor.shutdown
+        return nil
+      end
+
+      # Begin an immediate shutdown. In-progress tasks will be allowed to
+      # complete but enqueued tasks will be dismissed and no new tasks
+      # will be accepted. Has no additional effect if the thread pool is
+      # not running.
+      def kill
+        @executor.shutdownNow
+        nil
+      end
+
+      protected
+
+      def set_shutdown_hook
+        # without this the process may fail to exit
+        at_exit { self.kill }
+      end
+    end
+  end
 end

lib/concurrent/executor/java_single_thread_executor.rb

@@ -5,85 +5,16 @@ module Concurrent
 
     # @!macro single_thread_executor
     class JavaSingleThreadExecutor
-      include Executor
+      include JavaExecutor
 
       # Create a new thread pool.
       #
       # @see http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
       # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
       def initialize(opts = {})
-
         @executor = java.util.concurrent.Executors.newSingleThreadExecutor
-
-        # without this the process may fail to exit
-        at_exit { self.kill }
-      end
-
-      # Is the thread pool running?
-      #
-      # @return [Boolean] `true` when running, `false` when shutting down or shutdown
-      def running?
-        ! (@executor.isShutdown || @executor.isTerminated)
-      end
-
-      # Is the thread pool shutdown?
-      #
-      # @return [Boolean] `true` when shutdown, `false` when shutting down or running
-      def shutdown?
-        @executor.isShutdown
-      end
-
-      # Block until thread pool shutdown is complete or until `timeout` seconds have
-      # passed.
-      #
-      # @note Does not initiate shutdown or termination. Either `shutdown` or `kill`
-      #   must be called before this method (or on another thread).
-      #
-      # @param [Integer] timeout the maximum number of seconds to wait for shutdown to complete
-      #
-      # @return [Boolean] `true` if shutdown complete or false on `timeout`
-      def wait_for_termination(timeout)
-        @executor.awaitTermination(timeout.to_i, java.util.concurrent.TimeUnit::SECONDS)
-      end
-
-      # Submit a task to the thread pool for asynchronous processing.
-      #
-      # @param [Array] args zero or more arguments to be passed to the task
-      #
-      # @yield the asynchronous task to perform
-      #
-      # @return [Boolean] `true` if the task is queued, `false` if the thread pool
-      #   is not running
-      #
-      # @raise [ArgumentError] if no task is given
-      def post(*args)
-        raise ArgumentError.new('no block given') unless block_given?
-        if running?
-          @executor.submit{ yield(*args) }
-          true
-        else
-          false
-        end
-      rescue Java::JavaUtilConcurrent::RejectedExecutionException => ex
-        raise RejectedExecutionError
-      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
-        @executor.shutdown
-        return nil
-      end
-
-      # Begin an immediate shutdown. In-progress tasks will be allowed to
-      # complete but enqueued tasks will be dismissed and no new tasks
-      # will be accepted. Has no additional effect if the thread pool is
-      # not running.
-      def kill
-        @executor.shutdownNow
-        return nil
+        set_shutdown_hook
       end
     end
   end

lib/concurrent/executor/java_thread_pool_executor.rb

@@ -5,7 +5,7 @@ module Concurrent
 
     # @!macro thread_pool_executor
     class JavaThreadPoolExecutor
-      include Executor
+      include JavaExecutor
 
       # Default maximum number of threads that will be created in the pool.
       DEFAULT_MAX_POOL_SIZE = java.lang.Integer::MAX_VALUE # 2147483647
@@ -86,8 +86,7 @@ def initialize(opts = {})
           idletime, java.util.concurrent.TimeUnit::SECONDS,
           queue, OVERFLOW_POLICIES[@overflow_policy].new)
 
-        # without this the process may fail to exit
-        at_exit { self.kill }
+        set_shutdown_hook
       end
 
       # The minimum number of threads that may be retained in the pool.
@@ -168,68 +167,16 @@ def status
       #
       # @return [Boolean] `true` when running, `false` when shutting down or shutdown
       def running?
-        ! (@executor.isShutdown || @executor.isTerminated || @executor.isTerminating)
-      end
-
-      # Is the thread pool shutdown?
-      #
-      # @return [Boolean] `true` when shutdown, `false` when shutting down or running
-      def shutdown?
-        @executor.isShutdown
-      end
-
-      # Block until thread pool shutdown is complete or until `timeout` seconds have
-      # passed.
-      #
-      # @note Does not initiate shutdown or termination. Either `shutdown` or `kill`
-      #   must be called before this method (or on another thread).
-      #
-      # @param [Integer] timeout the maximum number of seconds to wait for shutdown to complete
-      #
-      # @return [Boolean] `true` if shutdown complete or false on `timeout`
-      def wait_for_termination(timeout)
-        @executor.awaitTermination(timeout.to_i, java.util.concurrent.TimeUnit::SECONDS)
-      end
-
-      # Submit a task to the thread pool for asynchronous processing.
-      #
-      # @param [Array] args zero or more arguments to be passed to the task
-      #
-      # @yield the asynchronous task to perform
-      #
-      # @return [Boolean] `true` if the task is queued, `false` if the thread pool
-      #   is not running
-      #
-      # @raise [ArgumentError] if no task is given
-      def post(*args)
-        raise ArgumentError.new('no block given') unless block_given?
-        if running?
-          @executor.submit{ yield(*args) }
-          true
-        else
-          false
-        end
-      rescue Java::JavaUtilConcurrent::RejectedExecutionException => ex
-        raise RejectedExecutionError
+        super && ! @executor.isTerminating
       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
-        @executor.shutdown
-        @executor.getQueue.clear
-        return nil
-      end
-
-      # Begin an immediate shutdown. In-progress tasks will be allowed to
-      # complete but enqueued tasks will be dismissed and no new tasks
-      # will be accepted. Has no additional effect if the thread pool is
-      # not running.
-      def kill
-        @executor.shutdownNow
+        super
         @executor.getQueue.clear
-        return nil
+        nil
       end
     end
   end