   (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)))))
+     (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 []

(defrecord ElasticsearchWrite []
  p-ext/Pipeline
  (read-batch
    [_ event]
    (function/read-batch event))

  (write-batch
    [_ {results :onyx.core/results
        connection :elasticsearch/connection
        defaults :elasticsearch/doc-defaults}]
    (doseq [msg (mapcat :leaves (:tree results))]
      (let [document (or (:elasticsearch/message (:message msg)) (:message msg))
            settings (if
                       (or (= :delete (:elasticsearch/write-type defaults))
                           (contains-some? (:message msg) :elasticsearch/message :elasticsearch/write-type))
                       (merge defaults (select-keys
                                         (:message msg) [:elasticsearch/index
                                                         :elasticsearch/doc-id
                                                         :elasticsearch/mapping
                                                         :elasticsearch/write-type]))
                       defaults)]
        (log/debug (str "Message Settings: " settings))
        (write-elasticsearch connection document settings)))
    {})

  (seal-resource
    [_ _]
    {}))

(defn write-messages [_]

      (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
