   ([opts :- OptsElasticsearch]
    (let [BUFSIZE (-> opts :bulk-size)
          TIMEOUT_MS (-> opts :bulk-timeout-ms)
-         ch (chan BUFSIZE)
-         proc (go-loop [buf []]
+         ch (async/chan BUFSIZE)
+         proc (async/go-loop [buf []]
                 (let [timed-out (async/timeout TIMEOUT_MS)
-                      [x select] (alts! [ch timed-out])
+                      [x select] (async/alts! [ch timed-out])
                       newbuf
                       (cond
                         ;; got nil from the message chan, index buf,

(ns runbld.store
  (:refer-clojure :exclude [get])
  (:require [runbld.schema :refer :all]
            [schema.core :as s])
  (:require [clojure.core.async :as async
             :refer [go go-loop chan >! <! alts! close!]]
            [clojure.string :as str]
            [elasticsearch.document :as doc]
            [elasticsearch.indices :as indices]
            [elasticsearch.connection :as conn]
            [elasticsearch.connection.http :as http]
            [runbld.util.date :as date]
            [runbld.io :as io]
            [runbld.vcs :refer [VcsRepo]]
            [slingshot.slingshot :refer [try+ throw+]])
  (:import (elasticsearch.connection Connection)))


        t (name DocType)
        id (:id d)
        es-addr {:index idx :type t :id id}]
    (debug/log "Saving doc in ES:" d)
    (doc/update conn idx t id {:body {:doc d
                                      :doc_as_upsert true}
                               :query-params {:refresh true}})
