core async pool, go block vs threads, etc

core_async_pool_test.clj

  (require
    '[clojure.core.async :as async]
    '[clojure.core.async.impl.protocols :as protocols]
    '[clojure.core.async.impl.concurrent :as conc]
    '[clojure.core.async.impl.exec.threadpool :as tp])

  (import 'java.util.concurrent.Executors)

  (def  executor-svc (Executors/newFixedThreadPool
                       4
                       (conc/counted-thread-factory "my-async-dispatch-%d" true)))
  
  (def my-executor
    (reify protocols/Executor
      (protocols/exec [this r]
        (.execute executor-svc ^Runnable r))))

  (alter-var-root #'clojure.core.async.impl.dispatch/executor
    (constantly (delay my-executor)))

  (defn- thread-id []
    (hash (Thread/currentThread)))
  
  (defn net-call [& args]
    (println "Go Thread" (thread-id) "Starting" (first args))
    (println "Pool size:" (.getPoolSize executor-svc) "Queue size:" (.. executor-svc getQueue size))
    (println "Go Thread" (thread-id) "Ending" (first args))
    (slurp "http://localhost:8000/ping") ;; Network IO with a slow server
    ;; (throw (ex-info "error from go" {})) ;; uncomment to test exception in go block
    :ok)

  (time
    (run! async/<!!
      (mapv
        #(async/go (net-call %))
        (range 30))))

  (time
    (run!
      async/<!!
      (mapv
        #(async/thread
           (println "Thread" (thread-id) "Starting" %)
           (net-call)
           (println "Thread" (thread-id) "Ending" %))
        (range 30))))


  (time
    (let [orgs-channel (async/to-chan! (range 30))
          response-channel (async/chan (async/dropping-buffer 10))]
      (async/pipeline-async 1 response-channel net-call orgs-channel)
      (async/<!! (async/go-loop [response (async/<! response-channel)]
                   (if response
                     (recur (async/<! response-channel))
                     {:message "complete"})))))