(defonce ^:dynamic ^:private *shared* (atom {}))

(def ^:private +units+)

(defn ->timeunit
  (^TimeUnit [obj]
   (cond (keyword? obj)
         (get +units+ obj)

         :else
         obj)))

(defn all-stacktraces
  ([]
   (Thread/getAllStackTraces)))

(defn aq:executor
  ([{:keys [target] :as opts}]
   (let [queue    (aq:new)
         executor (exe/executor:single 1)
         submit   (aq:submit executor queue opts)]
     {:queue queue
      :executor executor
      :submit submit
      :options opts})))

(defn aq:new
  ([]
   (atom [])))

(defn aq:process
  ([f aq max-batch]
   (loop []
     (let [queue (at/swap-return! aq
                                  (fn [queue] [queue []]))]
       (->> (partition-all max-batch queue)
            (mapv f))))))

(defn aq:submit
  ([executor queue {:keys [target handler max-batch interval] :as opts}]
   (let [bulk-fn (fn []
                   (aq:process (fn [items]
                                 (-> (handler target items)
                                     (env/explode)))
                               queue
                               max-batch))]
     (fn [& entries]
       (do (apply swap! queue conj entries)
           (exe/submit-notify executor
                              bulk-fn
                              interval)
           nil)))))

(defn bulk
  ([client thunk]
   (bulk client thunk {}))
  ([client thunk opts]
   (let [{:keys [async catch chain post] :as opts} (req:opts-init opts)
         [bulk current] (bulk-collect client thunk chain)
         received (f/on:all current
                            (fn [& arr]
                              (reduce h/call (apply vector arr) post)))
         received (cond-> received
                    catch (f/on:exception catch))
         final    (f/future:chain received chain)
         _ (cond bulk
                 (bulk-process client bulk opts)

                 :else
                 (swap! (get *current* client) conj final))]
     (req:return final (boolean (or (not bulk)
                                    (get *transact* client)
                                    async))))))

(defn bulk:inputs
  ([client thunk]
   (let [context (bulk-context)]
     (binding [*bulk*    (assoc *bulk* client context)
               *current* (assoc *current* client (atom []))]
       (thunk)
       (:inputs @(:cache context))))))

(defn bulk:map
  ([client f inputs]
   (bulk:map client f inputs {}))
  ([client f inputs opts]
   (bulk client
         (fn []
           (mapv (partial f client) inputs))
         opts)))

(defn bulk:transact
  ([client thunk]
   (bulk:transact client thunk {}))
  ([client thunk opts]
   (let [{:keys [async chain pre post] :as opts} (req:opts-init opts)
         state (volatile! nil)
         _     (reduce h/call nil pre)]
     (bulk client
           (fn []
             (let [res (transact client thunk)]
               (vreset! state res)))
           (req:opts-clean opts))
     (-> @state
         (f/on:all (fn [& arr]
                     (reduce h/call (apply vector arr) post)))
         (f/future:chain chain)
         (req:return (boolean (or (get *bulk* client)
                                  (get *transact* client)
                                  async)))))))

(defn bus
  ([] (bus nil))
  ([m]
   (-> (bus:create m)
       (component/start))))

(defn bus:all-ids
  ([{:keys [state] :as bus}]
   (let [{:keys [threads]} @state]
     (set (keys threads)))))

(defn bus:all-threads
  ([{:keys [state] :as bus}]
   (let [{:keys [threads]} @state]
     (collection/map-vals :thread threads))))

(defn bus:close
  ([bus id]
   (bus:send bus id {:op :exit})))

(defn bus:close-all
  ([bus]
   (collection/map-entries (fn [id]
                    [id (bus:close bus id)])
                  (bus:all-ids bus))))

(defn bus:create
  ([] (bus:create nil))
  ([m]
   (map->Bus (merge {:state   (atom {})
                     :results (atom {})
                     :output  (queue/queue)
                     :counters {:sent (atom 0)
                                :received (atom 0)}}
                    m))))

(defn bus:deregister
  ([bus]
   (bus:deregister bus (bus:get-id bus)))
  ([{:keys [state] :as bus} id]
   (swap! state (fn [m]
                  (let [{:keys [thread]} (get-in m [:threads id])]
                    (-> m
                        (update :threads dissoc id)
                        (update :lookup dissoc thread)))))))

(defn bus:get-count
  ([{:keys [state] :as bus}]
   (-> state deref :threads count)))

(defn bus:get-id
  ([bus]
   (bus:get-id bus (thread/thread:current)))
  ([{:keys [state]} thread]
   (let [{:keys [lookup]} @state]
     (get lookup thread))))

(defn bus:get-queue
  ([bus]
   (bus:get-queue bus (thread/thread:current)))
  ([{:keys [state] :as bus} obj]
   (let [{:keys [threads lookup]} @state]
     (:queue (if (instance? Thread obj)
               (get threads (get lookup obj))
               (get threads obj))))))

(defn bus:get-thread
  ([{:keys [state]} id]
   (let [{:keys [threads]} @state]
     (:thread (get threads id)))))

(defn bus:has-id?
  ([{:keys [state]} id]
   (let [{:keys [threads]} @state]
     (boolean (get threads id)))))

(defn bus:kill
  ([bus id]
   (let [thread (bus:get-thread bus id)]
     (if (and thread (not= thread (thread/thread:current)))
       (doto thread
         (thread/thread:interrupt))))))

(defn bus:kill-all
  ([bus]
   (collection/map-entries (fn [id]
                    [id (bus:kill bus id)])
                  (bus:all-ids bus))))

(defn bus:open
  ([bus handler]
   (bus:open bus handler {}))
  ([bus handler {:keys [id] :as opts
                 :or {id (std.lib.foundation/sid)}}]
   (if-not (component/started? bus) (std.lib.foundation/error "Bus not started" {:bus bus}))
   
   (let [started  (f/incomplete)
         opts     (assoc opts :id id :started started :stopped (f/incomplete))
         _        (f/future
                    (run-handler bus handler opts))]
     started)))

(defn bus:register
  ([bus]
   (bus:register bus (std.lib.foundation/sid)))
  ([bus id]
   (bus:register bus id (thread/thread:current)))
  ([{:keys [state] :as bus} id thread]
   (let [queue (queue/queue)]
     (swap! state (fn [m]
                    (-> m
                        (assoc-in [:threads id]
                                  {:id id :thread thread :queue queue})
                        (assoc-in [:lookup thread] id))))
     queue)))

(defn bus:reset-counters
  [bus]
  (collection/map-vals #(atom/swap-return! % (fn [v] [v 0]))
              (:counters bus)))

(defn bus:send
  ([{:keys [results counters] :as bus} id msg]
   (if-let [queue  (bus:get-queue bus id)]
     (let [msg-id (or (:id msg) (std.lib.foundation/sid))
           return (f/incomplete)]
       (swap! results assoc msg-id return)
       (queue/put queue (assoc msg :id msg-id))
       (swap! (:sent counters) inc)
       return))))

(defn bus:send-all
  ([bus msg]
   (collection/map-entries (fn [id] [id (bus:send bus id msg)]) (bus:all-ids bus))))

(defn bus:wait
  ([bus]
   (bus:wait bus {}))
  ([bus {:keys [timeout timeunit]
         :or {timeunit :ms}}]
   (queue/take (bus:get-queue bus) timeout timeunit)))

(defmacro bus:with-temp
  ([var & body]
   `(let [~var   (bus)
          ~'_    (bus:register ~var)]
      (try ~@body
           (finally (bus:deregister ~var)
                    (stop-bus ~var))))))

(defn bus?
  ([obj]
   (instance? Bus obj)))

(defn ^LinkedBlockingDeque deque
  ([]
   (LinkedBlockingDeque.))
  ([& elements]
   (LinkedBlockingDeque. ^java.util.Collection elements)))

(defn deque?
  ([obj]
   (instance? BlockingDeque obj)))

(defn drain
  ([^BlockingQueue queue]
   (let [target (java.util.ArrayList. (count queue))]
     (.drainTo queue target)
     target))
  ([^BlockingQueue queue ^long max]
   (let [target (java.util.ArrayList. max)]
     (drain queue max target)
     target))
  ([^BlockingQueue queue ^long max ^java.util.Collection target]
   (.drainTo queue target max)
   target))

(defn exec:at-capacity?
  ([executor]
   (= (exec:current-active executor)
      (exec:pool-max executor))))

(defn exec:await-termination
  ([^ExecutorService service]
   (exec:await-termination service Long/MAX_VALUE))
  ([^ExecutorService service ^long ms]
   (.awaitTermination service ms TimeUnit/MILLISECONDS)))

(defn exec:current-active
  ([^ThreadPoolExecutor service]
   (.getActiveCount service)))

(defn exec:current-completed
  ([^ThreadPoolExecutor service]
   (.getCompletedTaskCount service)))

(defn exec:current-size
  ([^ThreadPoolExecutor service]
   (.getPoolSize service)))

(defn exec:current-submitted
  ([^ThreadPoolExecutor service]
   (.getTaskCount service)))

(defn ^BlockingQueue exec:get-queue
  ([service]
   (if (instance? ThreadPoolExecutor service)
     (.getQueue ^ThreadPoolExecutor service)
     (throw (ex-info "Cannot access queue" {:service service})))))

(defn exec:increase-capacity
  ([executor]
   (exec:increase-capacity executor (* 2 (exec:pool-max executor))))
  ([executor n]
   (let [max (exec:pool-max executor)]
     (if (< max n) (exec:pool-max executor n))
     (exec:pool-size executor n))))

(defn exec:keep-alive
  ([^ThreadPoolExecutor service]
   (.getKeepAliveTime service TimeUnit/MILLISECONDS))
  ([^ThreadPoolExecutor service length]
   (.setKeepAliveTime service length TimeUnit/MILLISECONDS)))

(defn exec:pool-max
  ([^ThreadPoolExecutor service]
   (.getMaximumPoolSize service))
  ([^ThreadPoolExecutor service size]
   (.setMaximumPoolSize service size)))

(defn exec:pool-size
  ([^ThreadPoolExecutor service]
   (.getCorePoolSize service))
  ([^ThreadPoolExecutor service size]
   (.setCorePoolSize service (int size))))

(defn ^BlockingQueue exec:queue
  ([]
   (exec:queue nil))
  ([arg]
   (cond (q/queue? arg) arg

         (nil? arg) (q/queue)

         (integer? arg) (q/queue:fixed arg)

         (map? arg)
         (cond (integer? (:size arg))
               (q/queue:fixed (:size arg))

               :else
               (q/queue))

         :else (throw (ex-info "Invalid input" {:input arg})))))

(defn exec:rejected-handler
  ([^ThreadPoolExecutor service]
   (.getRejectedExecutionHandler service))
  ([^ThreadPoolExecutor service f]
   (.setRejectedExecutionHandler service
                                 (reify RejectedExecutionHandler
                                   (rejectedExecution [_ task executor]
                                     (f task executor))))))

(defn exec:shutdown
  ([^ExecutorService service]
   (doto service
     (track/untrack)
     (.shutdown))))

(defn exec:shutdown-now
  ([^ExecutorService service]
   (doto service
     (track/untrack)
     (.shutdownNow))))

(defn exec:shutdown?
  ([^ExecutorService service]
   (.isShutdown service)))

(defn exec:terminated?
  ([^ExecutorService service]
   (.isTerminated service)))

(defn exec:terminating?
  ([^ThreadPoolExecutor service]
   (.isTerminating service)))

(defmulti executor
  :type)

(defn- executor-string
  [executor]
  (let [tag (if (instance? ScheduledThreadPoolExecutor executor)
              "scheduled"
              "raw")]
    (str "#" tag ".executor" (executor:info executor [:type :counter :current]))))

(defn ^ExecutorService executor:cached
  ([]
   (doto (Executors/newCachedThreadPool)
     (track/track))))

(defn executor:health
  ([executor]
   (if (exec:shutdown? executor)
     {:status :not-healthy}
     {:status :ok})))

(defn executor:info
  ([^ThreadPoolExecutor executor]
   (executor:info executor #{:type :running :current :counter :options}))
  ([^ThreadPoolExecutor executor k]
   (let [[return items]  (cond (keyword? k)
                               [k #{k}]

                               :else
                               [identity (set k)])
         queue  (.getQueue executor)]
     (cond-> {}
       (:type items)    (assoc :type (executor:type executor))
       (:running items) (assoc :running (not (exec:shutdown? executor)))
       (:current items) (assoc :current {:threads (exec:current-size executor)
                                         :active  (exec:current-active executor)
                                         :queued  (count queue)
                                         :terminated (exec:terminated? executor)})
       (:counter items) (assoc :counter {:submit   (exec:current-submitted executor)
                                         :complete (exec:current-completed executor)})
       (:options items) (assoc :options {:pool {:size (exec:pool-size executor)
                                                :max (exec:pool-max executor)
                                                :keep-alive (exec:keep-alive executor)}
                                         :queue {:remaining (q/remaining-capacity queue)
                                                 :total (count queue)}})
       :then return))))

(def executor:kill  exec:shutdown-now)

(defn ^ExecutorService executor:pool
  ([size max keep-alive]
   (executor:pool size max keep-alive (q/queue)))
  ([^long size ^long max ^long keep-alive size-or-queue]
   (doto (ThreadPoolExecutor. size max keep-alive
                              TimeUnit/MILLISECONDS (exec:queue size-or-queue))
     (track/track))))

(defn executor:props
  ([^ThreadPoolExecutor executor]
   {:pool {:size       {:get exec:pool-size
                        :set exec:pool-size}
           :max        {:get exec:pool-max
                        :set exec:pool-max}
           :keep-alive {:get exec:keep-alive
                        :set exec:keep-alive}}
    :rejected-handler  {:get exec:rejected-handler
                        :set exec:rejected-handler}}))

(defn ^ScheduledThreadPoolExecutor executor:scheduled
  ([^long size]
   (doto (ScheduledThreadPoolExecutor. size)
     (track/track))))

(defn executor:share
  ([id ^ExecutorService executor]
   (swap! *shared* (fn [m]
                     (if (or (get @f/*pools* id)
                             (get m id))
                       (throw (ex-info "Already exists" {:id id}))
                       (assoc m id executor))))))

(defn executor:shared
  ([]
   (merge @*shared* @f/*pools*)))

(defn ^ExecutorService executor:single
  ([]
   (executor:single (q/queue)))
  ([size-or-queue]
   (doto (ThreadPoolExecutor. 0 1 0 TimeUnit/MILLISECONDS (exec:queue size-or-queue))
     (track/track))))

(def executor:start  identity)

(def executor:started? (comp not exec:shutdown?))

(def executor:stop  exec:shutdown)

(def executor:stopped? exec:shutdown?)

(def executor:submit submit)

(defn executor:type
  ([^ThreadPoolExecutor executor]
   (cond (instance? ScheduledThreadPoolExecutor executor)
         :scheduled

         (= 1 (exec:pool-max executor))
         :single

         (and (= Integer/MAX_VALUE (exec:pool-max executor))
              (let [queue (exec:get-queue executor)]
                (and (zero? (q/remaining-capacity queue))
                     (zero? (count queue)))))
         :cached

         :else :pool)))

(defn executor:unshare
  ([id]
   (atom/swap-return! *shared* (fn [m] [(get m id) (dissoc m id)]))))

(defn hub-state
  [elems]
  {:ticket (f/incomplete)
   :queue  elems})

(defn hub:add-entries
  ([hub entries]
   (at/swap-return! hub
                   (fn [{:keys [ticket queue] :as m}]
                     (let [start (count queue)]
                       [[ticket start (count entries)] {:ticket ticket
                                                        :queue (into queue entries)}])))))

(defn hub:executor
  ([object {:keys [handler max-batch interval]
            :as opts}]
   (let [queue    (hub:new)
         executor (exe/executor:single 1)
         submit   (hub:submit object executor queue opts)]
     {:queue queue
      :executor executor
      :submit submit})))

(defn hub:new
  ([]
   (hub:new []))
  ([elems]
   (atom (hub-state elems))))

(defn hub:process
  ([f hub max-batch]
   (let [[ticket queue] (at/swap-return! hub
                                        (fn [{:keys [ticket queue]}]
                                          [[ticket queue] (hub-state [])]))]
     (->> (partition-all max-batch queue)
          (mapv f)
          (f/future:force ticket)))))

(defn hub:submit
  ([object executor hub {:keys [handler max-batch interval]}]
   (let [bulk-fn (fn [] (hub:process (fn [items]
                                       (handler object items))
                                     hub
                                     max-batch))]
     (fn [& entries]
       (let [[ticket start count] (hub:add-entries hub entries)
             hit  (exe/submit-notify executor
                                     bulk-fn
                                     interval)]
         [ticket start count hit])))))

(defn hub:wait
  [hub & [timeout]]
  (let [{:keys [ticket queue]} @hub]
    (if (not (zero? (count queue)))
      (deref ticket (or timeout 1000) :timeout))))

(defn peek
  ([^BlockingQueue queue]
   (.peek queue)))

(defn peek-first
  ([^BlockingDeque queue]
   (.peekFirst queue)))

(defn peek-last
  ([^BlockingDeque queue]
   (.peekLast queue)))

(defn pool
  ([{:keys [size max keep-alive poll resource] :as m}]
   (-> (pool:create m)
       (pool:start))))

(defn pool:acquire
  ([{:keys [state resource] :as pool}]
   (let [{:keys [thread-local]} resource
         resource-fn (fn [resource thread-id]
                       (-> resource
                           (update :used inc)
                           (assoc  :update-time (time/time-ns)
                                   :status :busy
                                   :thread-id thread-id)))
         update-fn   (fn [m thread-id id resource]
                       (-> m
                           (update :busy assoc id
                                   (resource-fn resource thread-id))
                           (update-in [:stats :acquire] inc)
                           (update-in [:lookup thread-id]
                                      (fnil #(conj % id) #{}))))
         acquire-fn  (fn [{:keys [busy idle options lookup] :as m}]
                       (let [{:keys [max]} options
                             [id resource] (first idle)
                             thread-id (t/thread:id)
                             local-id (first (get lookup thread-id))]
                         (cond (and thread-local local-id)
                               [[:success local-id
                                 (f/completed (:object (get busy local-id)))]
                                m]

                               id
                               [[:success id
                                 (f/completed (:object resource))]
                                (-> (update-fn m thread-id id resource)
                                    (update :idle dissoc id))]

                               (and max
                                    (<= max
                                        (+ (count busy)
                                           (count idle))))
                               [[:reject nil (ex-info "Maximum limit reached" {:max max})]
                                (update-in m [:stats :reject] inc)]

                               :else
                               [[:success (std.lib.foundation/sid) (f/incomplete)]
                                (update-in m [:stats :create] inc)])))
         [status id boxed] (atom/swap-return! state acquire-fn)
         allocate-fn (fn [state id boxed]
                       (f/fulfil boxed
                                 (fn []
                                   (let [resource (pool-resource id pool)]
                                     (swap! state (fn [m] (update-fn m (t/thread:id) id resource)))
                                     (:object resource)))))
         object  (if (f/future? boxed)
                   (do (if (not (f/future:complete? boxed))
                         (allocate-fn state id boxed))
                       @boxed)
                   boxed)]
     (case status
       :success [id object]
       :reject  (throw object)))))

(defn pool:cleanup
  ([{:keys [state] :as pool}]
   (let [current (time/time-ns)
         {:keys [stop]} (:resource pool)
         {:keys [health] :or {health (constantly {:status :ok})}} (:resource pool)
         {:keys [idle]} @state
         dead-ids (keep (fn [[id resource]]
                          (if-not (-> (health resource)
                                      (:status)
                                      (= :ok))
                            id))
                        idle)
         dispose (fn [m id] (dispose-fn m id stop))
         _    (swap! state (fn [m] (reduce dispose  m dead-ids)))
         cleanup-fn  (fn [{:keys [idle options] :as m}]
                       (let [{:keys [size keep-alive]} options
                             over (- (count idle) size)
                             [ids m] (cond (pos? over)
                                           (let [ids (->> idle
                                                          (keep (fn [[id {:keys [update-time]}]]
                                                                  (if (< (+ update-time keep-alive)
                                                                         current)
                                                                    id)))
                                                          (take over))]
                                             [ids (reduce dispose m ids)])

                                           :else
                                           [[] m])]
                         [ids (update-in m [:stats :cleanup] inc)]))
         ids  (atom/swap-return! state cleanup-fn)]
     ids)))

(defn pool:create
  ([m]
   (-> (assoc (apply dissoc m +opts+)
              :state (atom {:stats {:create  0
                                    :reject  0
                                    :dispose 0
                                    :acquire 0
                                    :release 0
                                    :history []
                                    :cleanup 0
                                    :total-time 0.0
                                    :busy-time  0.0}
                            :busy    {}
                            :idle    {}
                            :lookup  {}
                            :options (merge *defaults* (select-keys m +opts+))}))
       (map->Pool))))

(defn pool:dispose
  ([{:keys [state resource] :as pool} id]
   (swap! state dispose-fn id (:stop resource)) id))

(defn pool:dispose-over
  ([{:keys [state resource] :as pool} id]
   (atom/swap-return! state (fn [{:keys [idle options] :as m}]
                           (if (get idle id)
                             [id  (dispose-fn m id (:stop resource))]
                             [nil m])))))

(defn pool:dispose:mark
  ([]
   (vreset! *dispose* true)))

(defn pool:dispose:unmark
  ([]
   (vreset! *dispose* nil)))

(defn pool:release
  ([pool id]
   (pool:release pool id false))
  ([{:keys [state] :as pool} id dispose?]
   (let [resource-fn (fn [{:keys [update-time busy] :as resource}]
                       (let [current (time/time-ns)
                             diff (- current update-time)]
                         (assoc resource
                                :update-time current
                                :busy (+ busy diff)
                                :status :idle
                                :thread-id nil)))
         update-fn   (fn [m thread-id id resource]
                       (-> m
                           (update :idle assoc id (resource-fn resource))
                           (update :busy dissoc id)
                           (update-in [:stats :release] inc)
                           (update :lookup
                                   (fn [lu]
                                     (let [nentries (disj (get lu thread-id) id)]
                                       (if (empty? nentries)
                                         (dissoc lu thread-id)
                                         (assoc  lu thread-id nentries)))))))
         release-fn  (fn [{:keys [busy options] :as m}]
                       (let [{:keys [thread-id] :as resource} (get busy id)]
                         (cond (nil? resource)
                               [[:error (ex-info "Cannot find id" {:id id})]
                                m]

                               :else
                               [[:success id]
                                (update-fn m thread-id id resource)])))
         [[status id] {:keys [idle busy options executor]}] (atom/swap-return! state release-fn true)
         _  (if dispose? (pool:dispose pool id))
         _  (if (and executor
                     (< (:size options)
                        (+ (count idle)
                           (count busy))))
              (f/future:run (fn []
                              (pool:dispose-over pool id))
                            {:pool executor
                             :delay (:keep-alive options)}))]
     (case status
       :success id
       :error   nil))))

(defn pool:resources:busy
  ([{:keys [state] :as pool}]
   (collection/map-vals :object (:busy @state))))

(defn pool:resources:idle
  ([{:keys [state] :as pool}]
   (collection/map-vals :object (:idle @state))))

(defn pool:resources:thread
  ([pool]
   (pool:resources:thread pool (t/thread:current)))
  ([{:keys [state] :as pool} thread]
   (let [thread-id (t/thread:id thread)
         {:keys [busy lookup]} @state
         ids (get lookup thread-id)]
     (collection/map-juxt [identity (comp :object #(get busy %))] ids))))

(defmacro pool:with-resource
  ([[obj pool] & body]
   `((wrap-pool-resource
      (fn [obj#]
        (let [~obj obj#]
          ~@body))
      ~pool))))

(defn pool?
  ([obj]
   (instance? Pool obj)))

(defn pop
  ([^BlockingDeque queue]
   (.pop queue)))

(defn process-bulk
  ([f queue maximum]
   (loop [total  (count queue)]
     (let [n      (if (> total maximum)
                    maximum
                    total)
           txs (drain queue n)
           _   (when-not (empty? txs)
                 (f txs))]
       (let [more (count queue)]
         (if-not (zero? more)
           (recur more)))))))

(defn push
  ([^BlockingDeque queue element]
   (.push queue element)))

(defn put
  ([^BlockingQueue queue element]
   (put queue element nil))
  ([^BlockingQueue queue element timeout]
   (put queue element timeout :ms))
  ([^BlockingQueue queue element timeout timeunit]
   (if (nil? timeout)
     (.put queue element)
     (.offer queue element timeout (->timeunit timeunit)))))

(defn put-first
  ([^BlockingDeque queue element]
   (put-first queue element nil))
  ([^BlockingDeque queue element timeout]
   (put-first queue element timeout :ms))
  ([^BlockingDeque queue element timeout timeunit]
   (if (nil? timeout)
     (.putFirst queue element)
     (.offerFirst queue element timeout (->timeunit timeunit)))))

(defn put-last
  ([^BlockingDeque queue element]
   (put-last queue element nil))
  ([^BlockingDeque queue element timeout]
   (put-last queue element timeout :ms))
  ([^BlockingDeque queue element timeout timeunit]
   (if (nil? timeout)
     (.putLast queue element)
     (.offerLast queue element timeout (->timeunit timeunit)))))

(defn ^BlockingQueue queue
  ([]
   (LinkedBlockingQueue.))
  ([& elements]
   (LinkedBlockingQueue. ^java.util.Collection elements)))

(defn ^BlockingQueue queue:fixed
  ([size]
   (ArrayBlockingQueue. size)))

(defn ^LimitedQueue queue:limited
  ([size]
   (LimitedQueue. size)))

(defn queue?
  ([obj]
   (instance? BlockingQueue obj)))

(defn relay
  ([{:keys [type] :as m}]
   (-> (relay:create m)
       (component/start))))

(defn get-bus
  []
  (or *bus*
      (resource/res :hara/relay)))

(defn relay:create
  ([{:keys [id type] :as m}]
   (map->Relay (assoc m
                      :id (or id (f/sid))
                      :instance (atom nil)))))

(defn remaining-capacity
  ([^BlockingQueue queue]
   (.remainingCapacity queue)))

(defn remove
  ([^BlockingQueue queue obj]
   (.remove queue obj)))

(defmacro req
  ([client command]
   (let [context (-> (meta &form)
                     (assoc :ns (env/ns-sym)))]
     `(req-fn ~client ~command {:context (quote ~context)})))
  ([client command opts]
   (let [context (-> (meta &form)
                     (assoc :ns (env/ns-sym)))]
     `(req-fn ~client ~command (assoc ~opts :context (quote ~context))))))

(defn req-fn
  ([client command]
   (req-fn client command {}))
  ([client command opts]
   (let [bulk     (get *bulk* client)
         transact (get *transact* client)
         opts     (req:opts-init opts)
         opts     (if transact
                    (transact-prep client transact command opts)
                    opts)]
     (cond *inputs*
           (swap! *inputs* conj command)

           bulk
           (req:unit client command opts)

           :else
           (req:single client command opts)))))

(defn req:applicative
  ([m]
   (map->ReqApplicative m)))

(defmacro req:bulk
  ([[client opts] & body]
   (let [context (-> (meta &form)
                     (assoc :ns (env/ns-sym)
                            :type :bulk))]
     `(bulk ~client
            (fn []
              ~@body)
            (assoc ~opts :context (quote ~context))))))

;; generated at runtime, no single source form

;; generated at runtime, no single source form

(defn req:command
  ([m]
   (map->Command m)))

(defmacro req:in
  ([& body]
   `(binding [*inputs* (atom [])]
      ((fn [] ~@body))
      @*inputs*)))

(defn req:opts
  ([opts]
   (req:opts opts nil))
  ([opts {:keys [pre post chain] :as m}]
   (cond-> opts
     m     (merge (dissoc m :chain :post :pre))
     pre   (update :pre   #(vec (into % pre)))
     post  (update :post  (comp vec (partial concat post)))
     chain (update :chain (comp vec (partial concat chain))))))

(defn req:opts-clean
  ([opts]
   (dissoc opts :measure :time :debug :pre :post
           :chain :async :final :received :transact)))

;; generated at runtime, no single source form

;; generated at runtime, no single source form

(defn req:run
  ([command client args]
   (req:run command client args {}))
  ([{:keys [options process] :as command} client args opts]
   (let [input-fn  (or (:input options) f/NIL)
         output-fn (or (:output options) f/NIL)
         input     (format-input command args (merge (input-fn args) opts))
         out-fn    #(format-output command % (merge (output-fn args) opts))
         opts (-> (merge (:default options) opts)
                  (r/req:opts process)
                  (r/req:opts {:post [out-fn]}))]
     (run-request command client input opts))))

(defmacro req:transact
  ([[client opts] & body]
   (let [context (-> (meta &form)
                     (assoc :ns (env/ns-sym)
                            :type :transact))]
     `(bulk:transact ~client
                     (fn []
                       ~@body)
                     (assoc ~opts :context (quote ~context))))))

(defn schedule
  ([^ScheduledThreadPoolExecutor service ^Callable f ^long interval]
   (schedule service f interval nil))
  ([^ScheduledThreadPoolExecutor service ^Callable f ^long interval {:keys [min]}]
   (let [^Callable f (cond-> f
                       min (wrap-min-time min))]
     (.schedule service f interval TimeUnit/MILLISECONDS))))

(defn schedule:fixed-delay
  ([^ScheduledThreadPoolExecutor service ^Runnable f ^long delay]
   (schedule:fixed-delay service f delay nil))
  ([^ScheduledThreadPoolExecutor service ^Runnable f ^long delay {:keys [initial min]}]
   (let [^Runnable f (cond-> f
                       min (wrap-min-time min))]
     (.scheduleWithFixedDelay service f (or initial 0) delay TimeUnit/MILLISECONDS))))

(defn schedule:fixed-rate
  ([^ScheduledThreadPoolExecutor service ^Runnable f ^long interval]
   (schedule:fixed-rate service f interval nil))
  ([^ScheduledThreadPoolExecutor service ^Runnable f ^long interval {:keys [initial min]}]
   (let [^Runnable f (cond-> f
                       min (wrap-min-time min))]
     (.scheduleAtFixedRate service f (or initial 0) interval TimeUnit/MILLISECONDS))))

(defn send
  ([{:keys [instance] :as relay} msg]
   (let [{:keys [in out]}  (or @instance
                               (f/error "Relay not started"
                                        {:relay relay :msg msg}))
         {:keys [op line]
          :as msg}    (if (string? msg)
                        {:op :read-some
                         :line msg}
                        msg)
         op  (or (transport/+read-alias+ op)
                 (+read-ops+ op)
                 (+control-ops+ op)
                 (f/error "Op not valid." {:input op
                                           :available (set (concat +control-ops+
                                                                   +read-ops+))
                                           :alias transport/+read-alias+}))]
     (case op
       :kill     (relay-stop relay)
       :exit     (do (.close ^OutputStream (:raw out))
                     (bus/bus:close (:bus in) (:id in)))
       :flush    (transport/send-write-flush out)
       :raw      (transport/send-write-raw out line)
       :partial  (transport/send-write-line out line)
       (transport/send-command in out op line msg)))))

(defn stacktrace
  ([]
   (stacktrace (thread:current)))
  ([^Thread thread]
   (.getStackTrace thread)))

(defn submit
  ([^ExecutorService service ^Callable f]
   (submit service f nil))
  ([^ExecutorService service f {:keys [min max delay default] :as m}]
   (let [^Callable f (cond-> f
                       min (wrap-min-time min (or delay 0)))
         opts (cond-> {:pool service}
                max     (assoc :timeout max)
                default (assoc :default default)
                delay   (assoc :delay delay))]
     (f/future:run f opts))))

(defn submit-notify
  ([^ExecutorService service ^Callable f]
   (submit-notify service f nil))
  ([^ExecutorService service f {:keys [min max delay default] :as m}]
   (if (pos? (q/remaining-capacity (exec:get-queue service)))
     (h/suppress (submit service f m)))))

(defn take
  ([^BlockingQueue queue]
   (take queue nil))
  ([^BlockingQueue queue timeout]
   (take queue timeout :ms))
  ([^BlockingQueue queue timeout timeunit]
   (if (nil? timeout)
     (.take queue)
     (.poll queue timeout (->timeunit timeunit)))))

(defn take-first
  ([^BlockingDeque queue]
   (.takeFirst queue))
  ([^BlockingDeque queue timeout]
   (take-first queue timeout :ms))
  ([^BlockingDeque queue timeout timeunit]
   (if (nil? timeout)
     (.takeFirst queue)
     (.pollFirst queue timeout (->timeunit timeunit)))))

(defn take-last
  ([^BlockingDeque queue]
   (.takeLast queue))
  ([^BlockingDeque queue timeout]
   (take-last queue timeout :ms))
  ([^BlockingDeque queue timeout timeunit]
   (if (nil? timeout)
     (.takeLast queue)
     (.pollLast queue timeout (->timeunit timeunit)))))

(defn ^Thread thread
  ([{:keys [^Runnable handler ^bool daemon priority classloader uncaught name start]}]
   (cond-> (Thread. handler)
     name        (doto (.setName name))
     daemon      (doto (.setDaemon daemon))
     priority    (doto (.setPriority priority))
     classloader (doto (.setContextClassLoader classloader))
     uncaught    (doto (.setUncaughtExceptionHandler uncaught))
     start (doto (.start)))))

(defn thread:active-count
  ([]
   (Thread/activeCount)))

(defn thread:alive?
  ([^Thread thread]
   (.isAlive thread)))

(defn thread:all
  ([]
   (keys (all-stacktraces))))

(defn thread:all-ids
  ([]
   (set (map #(.getId ^Thread %) (thread:all)))))

(defn thread:classloader
  ([]
   (thread:classloader (thread:current)))
  ([^Thread thread]
   (.getContextClassLoader thread))
  ([^Thread thread loader]
   (.setContextClassLoader thread loader)))

(defn thread:current
  ([]
   (Thread/currentThread)))

(defn thread:daemon?
  ([]
   (thread:daemon? (thread:current)))
  ([^Thread thread]
   (.isDaemon thread)))

(defn thread:dump
  ([]
   (Thread/dumpStack)))

(defn thread:global-uncaught
  ([]
   (Thread/getDefaultUncaughtExceptionHandler))
  ([f]
   (Thread/setDefaultUncaughtExceptionHandler f)))

(defn thread:has-access?
  ([^Thread thread]
   (try
     (.checkAccess thread)
     true
     (catch SecurityException e
       false))))

(defn thread:has-lock?
  ([^Object lock]
   (Thread/holdsLock lock)))

(defn thread:id
  ([] (thread:id (thread:current)))
  ([^Thread thread]
   (.getId thread)))

(defn thread:interrupt
  ([^Thread thread]
   (.interrupt thread)))

(defn thread:interrupted?
  ([^Thread thread]
   (.isInterrupted thread)))

(defn thread:join
  ([]
   (thread:join (thread:current)))
  ([^Thread thread]
   (.join thread))
  ([^Thread thread ^long millis]
   (.join thread millis))
  ([^Thread thread millis nanos]
   (.join thread millis nanos)))

(defn thread:notify
  ([^Object lock]
   (locking lock (.notify lock))))

(defn thread:notify-all
  ([^Object lock]
   (locking lock (.notifyAll lock))))

(defn thread:run
  ([^Thread thread]
   (.run thread)))

(defn thread:sleep
  ([^long ms]
   (Thread/sleep ms)))

(defn thread:spin
  ([]
   (Thread/onSpinWait)))

(defn thread:start
  ([^Thread thread]
   (.start thread)))

(defn thread:uncaught
  ([]
   (thread:uncaught (thread:current)))
  ([^Thread thread]
   (.getUncaughtExceptionHandler thread))
  ([^Thread thread f]
   (.setUncaughtExceptionHandler thread f)))

(defn thread:wait-on
  ([^Object lock]
   (locking lock (.wait lock))))

(defn thread:yield
  ([]
   (Thread/yield)))

(defn transact
  ([client thunk]
   (let [top       (get *transact* client)
         context   (or top (transact-context))
         start-cmd (transact-start client)
         end-cmd   (transact-end client)
         _         (if (nil? top)
                     (req-fn client start-cmd {:async true}))
         results   (binding [*transact* (assoc *transact* client context)]
                     (thunk)
                     (:outputs @context))
         final     (-> @context :inputs :final)
         output    (if (nil? top)
                     (req-fn client end-cmd {:async true}))
         _   (if output
               (f/on:complete output
                              (fn [val err]
                                (if err
                                  (f/future:force final :exception err)
                                  (f/future:force final :value val)))))]
     results)))

(defn ^Callable wrap-min-time
  ([f total]
   (wrap-min-time f total 0))
  ([f total delay]
   (fn []
     (let [start (System/currentTimeMillis)
           _ (if (and delay (pos? delay)) (Thread/sleep (long delay)))
           _ (f)
           end (System/currentTimeMillis)
           duration (- end start)]
       (if (> total duration)
         (Thread/sleep (long (- total duration))))))))