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

  (let [query-list (if query [:query query] [])
        sort-list (if sort [:sort sort] [])]
    (->
      (cond
        (and index mapping) (run-as client-type :search conn index mapping query-list sort-list :from start-index :scroll scroll)
        (not (nil? index)) (run-as client-type :search-all-types conn index query-list sort-list :from start-index :scroll scroll)
        :else (run-as client-type :search-all-indexes-and-types conn query-list sort-list :from start-index :scroll scroll)))))

(defn- start-commit-loop! [write-chunk? commit-ch log k]
  (go-loop []

  (:require [onyx.peer.function :as function]
            [onyx.extensions :as extensions]
            [onyx.peer.pipeline-extensions :as p-ext]
            [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]
