 (defn send-iface [iface batches opts]
   ((whichfn iface) batches opts))
 
-(defn part [rdr opts]
-  (partition-all (:batch opts) (line-seq rdr)))
+(defn part [rdr {:keys [batch index type]}]
+  (if (and index type)
+    (let [bulk-header (json/encode {:index {:_index index :_type type}})]
+      (partition-all (* batch 2) (interleave (repeat bulk-header)
+                                             (line-seq rdr))))
+    (partition-all batch (line-seq rdr))))
 
 (defn main [{:keys [file offset-file interface batch reset] :as opts}]
   (with-offset [offset offset-file set-new-offset!]

  (:gen-class)
  (:require [clojure.tools.cli :as cli]
            [clojure.tools.logging :as log]
            [monocle.io :as io]
            [monocle.rabbitmq :as rmq])
  (:use [monocle.io :only [with-offset]]))

   ["--trustpw" "Trust keystore password"]
   ["--index" (str "Format the batches in elasticsearch's bulk\n"
                   "                                "
                   " format.  Also need --type.")]
   ["--type" "The type to provide to elasticsearch"]
   ["-h" "--help" "Help!" :default false :flag true]])

(defn amqp-opts [iface]
