rabbitmq.clj

(ns rabbitmq
  (:import (com.rabbitmq.client ConnectionParameters 
				ConnectionFactory
				QueueingConsumer)))

(defstruct connection-info
  :username :password :virtual-host :heartbeat :host :port)

(defn connect [info]
  (let [connection (.newConnection (ConnectionFactory. 
				    (doto (ConnectionParameters.)
				      (.setUsername (info :username))
				      (.setPassword (info :password))
				      (.setVirtualHost (info :virtual-host))
				      (.setRequestedHeartBeat (info :heartbeat))))
				   (info :host)
				   (info :port))
	channel (.createChannel connection)]
    [connection channel]))


(defn disconnect [connection channel]
  (.close channel)
  (.close connection))

(defn deliver [consumer channel]
  (let [delivery (.nextDelivery consumer)]
    (.basicAck channel (.. delivery getEnvelope getDeliveryTag) false)
    delivery))

(def default-connection (struct-map connection-info 
			  :username "guest" 
			  :password "guest"
			  :virtual-host "/"
			  :heartbeat 0
			  :host "localhost"
			  :port 5672))

(defn produce []
  (let [[conn channel] (connect default-connection)]
    (.queueDeclare channel "SimpleQueue")
    (.basicPublish channel "" "SimpleQueue" nil (.getBytes (str "the time is")))
    (disconnect conn channel)))

(defn consume []
  (let [[conn channel] (connect default-connection)
        consumer (QueueingConsumer. channel)]
    (.queueDeclare channel "SimpleQueue")
    (.basicConsume channel "SimpleQueue" consumer)
    (prn "Message: " (String. (.getBody (deliver consumer channel))))))