   "Forwards events to Graphite."
   (:refer-clojure :exclude [replace])
   (:import
-   (java.net Socket)
+   (java.net Socket
+             DatagramSocket
+             DatagramPacket
+             InetAddress)
    (java.io Writer
             OutputStreamWriter)
    (org.jboss.netty.channel Channels)


    (fn [event]
      (when (:metric event)
        (with-pool [[sock out] pool (:claim-timeout opts)]
                   (let [string (str (join " " [(path event)
                                                (float (:metric event))
                                                (int (:time event))])
                                     "\n")]
                     (.write ^OutputStreamWriter out string)
                     (.flush ^OutputStreamWriter out)))))))



        riemann.pool
        riemann.common))

(defprotocol GraphiteClient
  (open [client]
        "Creates a Graphite client")
  (send! [client line]
        "Sends a formatted line to Graphite")
  (close [client]
         "Cleans up"))

(defrecord GraphiteTCPClient [host port]
  GraphiteClient
  (open [this]
    (let [sock (Socket. host port)]
      (assoc this 
             :socket sock
             :out (OutputStreamWriter. (.getOutputStream sock)))))
  (send! [this line]
    (let [out (:out this)]
      (.write ^OutputStreamWriter out line)
      (.flush ^OutputStreamWriter out)))
  (close [this]
    (.close (:out this))
    (.close (:socket this))))

(defrecord GraphiteUDPClient [host port]
  GraphiteClient
  (open [this]
    (assoc this 
           :socket (DatagramSocket.)
           :host host
           :port port))
  (send! [this line]
    (let [bytes (.getBytes line)
          length (count line)
          addr (InetAddress/getByName (:host this))
          datagram (DatagramPacket. bytes length addr (:port this))]
      (.send (:socket this) datagram)))
  (close [this]
    (.close (:socket this))))

(defn graphite-path-basic
  "Constructs a path for an event. Takes the hostname fqdn, reversed,
