Merge pull request #210 from arnaudgeiser/master

Add new-thread-fn parameter to manifold.executor/thread-factory

Author: KingMob

src/manifold/executor.clj

@@ -32,25 +32,46 @@
        (finally
          (.set executor-thread-local executor#)))))
 
+(defn- ^Thread new-thread
+  "Creates a new `java.lang.Thread`.
+
+  It represents the default implementation on `thread-factory` when the
+  `new-thread-fn` argument is not passed.
+
+  Some libraries require a different implementation of a `java.lang.Thread`.
+  That's the case of Netty which behaves differently when
+  running on a `io.netty.util.concurrent.FastThreadLocalThread`."
+  [group target name stack-size]
+  (Thread. group target name stack-size))
+
 (defn ^ThreadFactory thread-factory
+  "Returns a `java.util.concurrent.ThreadFactory`.
+
+   |:---|:----
+   | `name-generator` | a zero-argument function, which, when invoked returns the name of the `java.lang.Thread` that will be created. |
+   | `executor-promise` | a promise eventually containing a `java.util.concurrent.Executor` that will be stored on `manifold.executor/executor-thread-local`. |
+   | `stack-size` | the desired stack size for the new thread, or nil/zero to indicate that this parameter is to be ignored. |
+   | `daemon?` | marks the created threads as either daemon or user threads. The Java Virtual Machine exits when the only threads running are all daemon threads. |
+   | `new-thread-fn` | a four arguments function which returns an implementation of `java.lang.Thread` when called. |"
   ([name-generator executor-promise]
-   (thread-factory name-generator executor-promise nil true))
+   (thread-factory name-generator executor-promise nil true nil))
   ([name-generator executor-promise stack-size]
-   (thread-factory name-generator executor-promise stack-size true))
+   (thread-factory name-generator executor-promise stack-size true nil))
   ([name-generator executor-promise stack-size daemon?]
-   (reify ThreadFactory
-     (newThread [_ runnable]
-       (let [name        (name-generator)
-             curr-loader (.getClassLoader (class thread-factory))
-             f           #(do
-                            (.set executor-thread-local @executor-promise)
-                            (.run ^Runnable runnable))]
-         (doto
-           (if stack-size
-             (Thread. nil f name stack-size)
-             (Thread. nil f name))
-           (.setDaemon daemon?)
-           (.setContextClassLoader curr-loader)))))))
+   (thread-factory name-generator executor-promise stack-size daemon? nil))
+  ([name-generator executor-promise stack-size daemon? new-thread-fn]
+   (let [new-thread (or new-thread-fn new-thread)]
+     (reify ThreadFactory
+       (newThread [_ runnable]
+         (let [name        (name-generator)
+               curr-loader (.getClassLoader (class thread-factory))
+               f           #(do
+                              (.set executor-thread-local @executor-promise)
+                              (.run ^Runnable runnable))
+               thread      ^Thread (new-thread nil f name (or stack-size 0))]
+           (doto thread
+             (.setDaemon daemon?)
+             (.setContextClassLoader curr-loader))))))))
 
 ;;;
 

test/manifold/executor_test.clj

@@ -40,3 +40,31 @@
                 (let [l (clojure.lang.RT/baseLoader)]
                   (deliver in-thread-loader l))))
     (is (instance? clojure.lang.DynamicClassLoader @in-thread-loader))))
+
+(defn- ^ThreadFactory thread-factory
+  ([] (thread-factory nil))
+  ([new-thread-fn] (thread-factory new-thread-fn nil))
+  ([new-thread-fn stack-size]
+   (let [num-threads (atom 0)
+         tf (e/thread-factory
+             #(str "my-pool-prefix" (swap! num-threads inc))
+             (deliver (promise) nil)
+             stack-size
+             false
+             new-thread-fn)]
+     tf)))
+
+(deftest test-thread-factory
+  (let [tf (thread-factory)]
+    (is (.newThread tf (constantly nil))))
+  (let [tf (thread-factory
+            (fn [group target _ stack-size]
+              (Thread. group target "custom-name" stack-size)))
+        thread (.newThread tf (constantly nil))]
+    (is (= "custom-name" (.getName thread))))
+  (let [tf (thread-factory
+            (fn [group target _ stack-size]
+              (Thread. group target "custom-name" stack-size))
+            500)
+        thread (.newThread tf (constantly nil))]
+    (is (= "custom-name" (.getName thread)))))