 (defrecord ElasticsearchWrite []
   p-ext/Pipeline
   (read-batch
-    [_ event]
+      [_ event]
     (function/read-batch event))
 
   (write-batch
-    [_ {results :onyx.core/results
-        connection :elasticsearch/connection
-        defaults :elasticsearch/doc-defaults}]
+      [_ {results :onyx.core/results
+          connection :elasticsearch/connection
+          default-vals :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)]
+                         (or (= :delete (:elasticsearch/write-type default-vals))
+                             (contains-some? (:message msg) :elasticsearch/message :elasticsearch/write-type))
+                       (merge default-vals (select-keys
+                                            (:message msg) [:elasticsearch/index
+                                                            :elasticsearch/doc-id
+                                                            :elasticsearch/mapping
+                                                            :elasticsearch/write-type]))
+                       default-vals)]
         (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

                              read-ch retry-ch commit-ch]
  p-ext/Pipeline
  (write-batch
      [_ event]
    (function/write-batch event))

  (read-batch [_ _]
