              DatagramSocket
              DatagramPacket
              InetAddress)
-   (java.io Writer OutputStreamWriter))
+   (java.io Writer OutputStreamWriter BufferedWriter))
   (:use [clojure.string :only [split join replace]]
         clojure.tools.logging
         riemann.pool

  (let [opts (merge {:host "127.0.0.1"
                     :port 9999
                     :protocol :tcp
                     :tls false
                     :claim-timeout 0.1
                     :pool-size 4} opts)

        pool (fixed-pool
               (fn []
                 (info "Connecting to " (select-keys opts [:host :port :security.protocol :tls]))
                 (let [
                       host (:host opts)
                       port (:port opts)
                       client (open (condp = (:protocol opts)
                                      :tcp (LogStashTCPClient. host port opts)
                                      :udp (LogStashUDPClient. host port)))]
                   (info "Connected")
                   client))
               (fn [client]

  (close [client]
         "Cleans up (closes sockets etc.)"))

(defrecord LogStashTLSClient [^String host ^int port opts]
  LogStashClient
  (open [this]
    (let [sock  (socket  (ssl-context (:key opts) (:cert opts) (:ca-cert opts))
                host
                port)]
          (.startHandshake sock)
          (assoc this
                 :socket sock
                 :out (BufferedWriter. (OutputStreamWriter. (.getOutputStream sock))))))
  (send-line [this line]
    (let [out (:out this)]
          (.write ^BufferedWriter out  ^String line)
          (.flush ^BufferedWriter out)))
  (close [this]
    (.close ^BufferedWriter (:out this))
    (.close ^Socket (:socket this))))

(defrecord LogStashTCPClient [^String host ^int port]
  LogStashClient
  (open [this]
    (let [sock (Socket. host port)]
      (assoc this
             :socket sock
             :out (OutputStreamWriter. (.getOutputStream sock)))))
  (send-line [this line]
    (let [out (:out this)]
      (.write ^OutputStreamWriter out ^String line)
      (.flush ^OutputStreamWriter out)))
  (close [this]
    (.close ^OutputStreamWriter (:out this))
    (.close ^Socket (:socket this))))

(defrecord LogStashUDPClient [^String host ^int port]
  LogStashClient
