Switch to use thread factory from manifold.executor (#426)

Author: kachayev

project.clj

@@ -12,7 +12,7 @@
 
 (def other-dependencies
   '[[org.clojure/tools.logging "0.4.1" :exclusions [org.clojure/clojure]]
-    [manifold "0.1.9-alpha2"]
+    [manifold "0.1.9-SNAPSHOT"]
     [byte-streams "0.2.5-alpha2"]
     [potemkin "0.4.5"]])
 

src/aleph/netty.clj

@@ -4,6 +4,7 @@
     [byte-streams :as bs]
     [clojure.tools.logging :as log]
     [manifold.deferred :as d]
+    [manifold.executor :as e]
     [manifold.stream :as s]
     [manifold.stream.core :as manifold]
     [primitive-math :as p]
@@ -51,12 +52,16 @@
      ResourceLeakDetector$Level]
     [java.net URI SocketAddress InetSocketAddress]
     [io.netty.util.concurrent
-     GenericFutureListener Future DefaultThreadFactory]
+     GenericFutureListener Future]
     [java.io InputStream File]
     [java.nio ByteBuffer]
     [io.netty.util.internal SystemPropertyUtil]
     [java.util.concurrent
-     ConcurrentHashMap CancellationException ScheduledFuture TimeUnit]
+     ConcurrentHashMap
+     CancellationException
+     ScheduledFuture
+     TimeUnit
+     ThreadFactory]
     [java.util.concurrent.atomic
      AtomicLong]
     [io.netty.util.internal.logging
@@ -721,18 +726,26 @@
   (let [cpu-count (->> (Runtime/getRuntime) (.availableProcessors))]
     (max 1 (SystemPropertyUtil/getInt "io.netty.eventLoopThreads" (* cpu-count 2)))))
 
+(defn ^ThreadFactory enumerating-thread-factory [prefix daemon?]
+  (let [num-threads (atom 0)]
+    (e/thread-factory
+     #(str prefix "-" (swap! num-threads inc))
+     (deliver (promise) nil)
+     nil
+     daemon?)))
+
 (def ^String client-event-thread-pool-name "aleph-netty-client-event-pool")
 
 (def epoll-client-group
   (delay
     (let [thread-count (get-default-event-loop-threads)
-          thread-factory (DefaultThreadFactory. client-event-thread-pool-name true)]
+          thread-factory (enumerating-thread-factory client-event-thread-pool-name true)]
       (EpollEventLoopGroup. (long thread-count) thread-factory))))
 
 (def nio-client-group
   (delay
     (let [thread-count (get-default-event-loop-threads)
-          thread-factory (DefaultThreadFactory. client-event-thread-pool-name true)]
+          thread-factory (enumerating-thread-factory client-event-thread-pool-name true)]
       (NioEventLoopGroup. (long thread-count) thread-factory))))
 
 (defn convert-address-types [address-types]
@@ -913,7 +926,7 @@
    epoll?]
   (let [num-cores      (.availableProcessors (Runtime/getRuntime))
         num-threads    (* 2 num-cores)
-        thread-factory (DefaultThreadFactory. "aleph-netty-server-event-pool" false)
+        thread-factory (enumerating-thread-factory "aleph-netty-server-event-pool" false)
         closed?        (atom false)
 
         ^EventLoopGroup group