   :block-start          Wait for the pool's initial connections to open
                         before returning.
 
-  :protocol             Protocol to use. Either :tcp (default) or :udp.
+  :protocol             Protocol to use. Either :tcp (default), :tls or :udp.
 
   TLS options:
-  :tls?             Whether to enable TLS
   :key              A PKCS8-encoded private key file
   :cert             The corresponding public certificate
   :ca-cert          The certificate of the CA which signed this key"

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

(defn get-socket
  [host  port  opts]
  (cond
     (= ^Boolean (:tls opts) true)
         (socket  (ssl-context (:key opts) (:cert opts) (:ca-cert opts))
                              host
                              port)
     :else
          (Socket. host port)))

(defrecord LogStashTCPClient [^String host ^int port opts]
  LogStashClient

  (open [this]

    (let [sock (get-socket host port opts)]

        (cond
           (= ^Boolean (:tls opts) true)
              (do
                (.getSession sock)
                (assoc this
                       :socket sock
                       :out (.getOutputStream sock))
              )
          :else
              (assoc this
                     :socket sock
                     :out (OutputStreamWriter. (.getOutputStream sock))))))

  (send-line [this line]
    (let [out (:out this)]
          (cond
            (= ^Boolean (:tls opts) true)
               (do
                (.write  out  (.getBytes (str line ) ))
                (.flush out))
            :else
              (do
                (.write ^OutputStreamWriter out ^String line)
                (.flush ^OutputStreamWriter out)))))

  (close [this]
    (cond
      (= ^Boolean (:tls opts) true)
         (do
            (.close (:out this))
            (.close ^Socket (:socket this)))
        :else
        (do
           (.close ^OutputStreamWriter (:out this))
           (.close ^Socket (:socket this))))))

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

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

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