Import backend code inside the repository.

backend/vendor/executors/core.clj

@@ -0,0 +1,176 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+;;
+;; Copyright (c) 2016 Andrey Antukh <niwi@niwi.nz>
+
+(ns executors.core
+  "A executos service abstraction layer."
+  (:import java.util.function.Supplier
+           java.util.concurrent.ForkJoinPool
+           java.util.concurrent.Future
+           java.util.concurrent.CompletableFuture
+           java.util.concurrent.ExecutorService
+           java.util.concurrent.TimeoutException
+           java.util.concurrent.ThreadFactory
+           java.util.concurrent.TimeUnit
+           java.util.concurrent.ScheduledExecutorService
+           java.util.concurrent.Executors))
+
+(def ^:const +max-priority+ Thread/MAX_PRIORITY)
+(def ^:const +min-priority+ Thread/MIN_PRIORITY)
+(def ^:const +norm-priority+ Thread/NORM_PRIORITY)
+
+;; --- Protocols
+
+(defprotocol IExecutor
+  (^:private -execute [_ task] "Execute a task in a executor.")
+  (^:private -submit [_ task] "Submit a task and return a promise."))
+
+(defprotocol IScheduledExecutor
+  (^:provate -schedule [_ ms task] "Schedule a task to execute in a future."))
+
+(defprotocol IScheduledTask
+  "A cancellation abstraction."
+  (-cancel [_])
+  (-cancelled? [_]))
+
+;; --- Implementation
+
+(defn- thread-factory-adapter
+  "Adapt a simple clojure function into a
+  ThreadFactory instance."
+  [func]
+  (reify ThreadFactory
+    (^Thread newThread [_ ^Runnable runnable]
+      (func runnable))))
+
+(defn- thread-factory
+  [{:keys [daemon priority]
+    :or {daemon true
+         priority Thread/NORM_PRIORITY}}]
+  (thread-factory-adapter
+   (fn [runnable]
+     (let [thread (Thread. ^Runnable runnable)]
+       (.setDaemon thread daemon)
+       (.setPriority thread priority)
+       thread))))
+
+(defn- resolve-thread-factory
+  [opts]
+  (cond
+    (map? opts) (thread-factory opts)
+    (fn? opts) (thread-factory-adapter opts)
+    (instance? ThreadFactory opts) opts
+    :else (throw (ex-info "Invalid thread factory" {}))))
+
+(deftype ScheduledTask [^Future fut]
+  clojure.lang.IDeref
+  (deref [_]
+    (.get fut))
+
+  clojure.lang.IBlockingDeref
+  (deref [_ ms default]
+    (try
+      (.get fut ms TimeUnit/MILLISECONDS)
+        (catch TimeoutException e
+          default)))
+
+    clojure.lang.IPending
+    (isRealized [_] (and (.isDone fut)
+                         (not (.isCancelled fut))))
+
+    IScheduledTask
+    (-cancelled? [_]
+      (.isCancelled fut))
+
+    (-cancel [_]
+      (when-not (.isCancelled fut)
+        (.cancel fut true))))
+
+(extend-type ExecutorService
+  IExecutor
+  (-execute [this task]
+    (CompletableFuture/runAsync ^Runnable task this))
+
+  (-submit [this task]
+    (let [supplier (reify Supplier (get [_] (task)))]
+      (CompletableFuture/supplyAsync supplier this))))
+
+(extend-type ScheduledExecutorService
+  IScheduledExecutor
+  (-schedule [this ms func]
+    (let [fut (.schedule this func ms TimeUnit/MILLISECONDS)]
+      (ScheduledTask. fut))))
+
+;; --- Public Api (Pool Constructors)
+
+(defn common-pool
+  "Get the common pool."
+  []
+  (ForkJoinPool/commonPool))
+
+(defn cached
+  "A cached thread pool constructor."
+  ([]
+   (Executors/newCachedThreadPool))
+  ([opts]
+   (let [factory (resolve-thread-factory opts)]
+     (Executors/newCachedThreadPool factory))))
+
+(defn fixed
+  "A fixed thread pool constructor."
+  ([n]
+   (Executors/newFixedThreadPool (int n)))
+  ([n opts]
+   (let [factory (resolve-thread-factory opts)]
+     (Executors/newFixedThreadPool (int n) factory))))
+
+(defn single-thread
+  "A single thread pool constructor."
+  ([]
+   (Executors/newSingleThreadExecutor))
+  ([opts]
+   (let [factory (resolve-thread-factory opts)]
+     (Executors/newSingleThreadExecutor factory))))
+
+(defn scheduled
+  "A scheduled thread pool constructo."
+  ([] (Executors/newScheduledThreadPool (int 1)))
+  ([n] (Executors/newScheduledThreadPool (int n)))
+  ([n opts]
+   (let [factory (resolve-thread-factory opts)]
+     (Executors/newScheduledThreadPool (int n) factory))))
+
+;; --- Public Api (Task Execution)
+
+(defn execute
+  "Execute a task in a provided executor.
+
+  A task is a plain clojure function or
+  jvm Runnable instance."
+  ([task]
+   (-> (common-pool)
+       (-execute task)))
+  ([executor task]
+   (-execute executor task)))
+
+(defn submit
+  "Submit a task to be executed in a provided executor
+  and return a promise that will be completed with
+  the return value of a task.
+
+  A task is a plain clojure function."
+  ([task]
+   (-> (common-pool)
+       (-submit task)))
+  ([executor task]
+   (-submit executor task)))
+
+(defn schedule
+  "Schedule task exection for some time in the future."
+  ([ms task]
+   (-> (common-pool)
+       (-schedule ms task)))
+  ([executor ms task]
+   (-schedule executor ms task)))