   (:require [onyx.peer.function :as function]
             [onyx.extensions :as extensions]
             [onyx.peer.pipeline-extensions :as p-ext]
-            [onyx.static.default-vals :refer [defaults arg-or-default]]
+            [onyx.static.default-vals :refer [default-vals arg-or-default]]
             [onyx.types :as t]
             [clojure.core.async :refer [chan go timeout <!! >!! alts!! sliding-buffer go-loop close! poll! offer!]]
             [clojurewerkz.elastisch.native  :as es]

(defn- create-es-client
  [client-type protocol host port cluster-name http-ops]
  (if
    (= client-type :http)
    (esr/connect (str (name protocol) "://" host ":" port) http-ops)
    (es/connect [[host port]] {"cluster.name" cluster-name})))


      (swap! pending-messages dissoc segment-id)))

  (retry-segment
      [_ _ segment-id]
    (when-let [msg (get @pending-messages segment-id)]
      (swap! pending-messages dissoc segment-id)
      (>!! retry-ch (t/input (java.util.UUID/randomUUID)
                             (:message msg)))))

  (pending?
      [_ _ segment-id]
    (get @pending-messages segment-id))

  (drained?
      [_ _]
    @drained?))

(defn read-messages
