concurrent.clj

(ns scraper.concurrent
  (:import [java.util.concurrent
              ExecutorService
              ExecutorCompletionService
              Executors
              CompletionService Future]))

(defn prln [& more]
  (.write
    *out*
    (str (clojure.string/join " " more) "\n"))
  (flush))

(defn take-seq
  "accepts an ExecutorCompletionService, and returns
  a lazy seq of values taken from the service"
  [worker]
  (lazy-seq
     (let [result (.take worker)]
       (cons (.get result)
             (take-seq worker)))))

(defn cmap
  "executes function f against coll in thread pool of size threads"
  [f threads coll]
  (let [pool (Executors/newFixedThreadPool threads)
        worker (ExecutorCompletionService. pool)]
    (doseq [x coll] (.submit worker #(f x)))
    (take (count coll) (take-seq worker))))

;; example for cmap
(defn go []
  (cmap
    (fn [x]
      (Thread/sleep (rand-int 1000))
      (prln "RUN" x)
      (str "DID" x))
    ;; 5 threads
    5
    ;; 20 item coll
    (range 20)))