 (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!]

   ["--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]

(defn delete-trailing-whitespace [s]
  (.replaceAll s "[ ]+\n" "\n"))

(defn mixmatch [& args]
  (= 2 (count (set (map boolean args)))))

(defn exit [cont msg]
  (println msg)
  (reset! cont false))

(defn fail [cont err]
  (log/fatalf err)
  (reset! cont false))

(defn -main [& args]
  (let [[{:keys [interface file offset-file
                 batch help index type] :as opts} args helpstr]
        (apply cli/cli args options)
        opts (assoc opts :ssl (select-keys opts [:client :clientpw
                                                 :trust :trustpw]))
        continue? (atom true)]
    (when-not offset-file
      (fail continue? "-o not supplied"))
    (when-not file
      (fail continue? "-f not supplied"))
    (when (mixmatch index type)
      (fail continue? "supply both --index and --type if you want ES bulk"))
    (when help
      (exit continue? (delete-trailing-whitespace helpstr)))
    (when @continue?
      (main opts))))
