Clojure JVM Kaazing JMS STOMP websocket client

websocket.clj

(ns apiclient.websocket
  (:import (java.net URL)
           (java.util Properties)
           (javax.naming Context InitialContext)
           (javax.jms BytesMessage Connection ConnectionFactory Destination
                      ExceptionListener JMSException
                      Message MessageConsumer MessageListener MessageProducer
                      Session TextMessage Topic TopicSubscriber)
           (com.kaazing.gateway.jms.client.stomp StompInitialContextFactory
                                                 GenericBaseMessage GenericBaseMessageImpl
                                                 GenericMessage GenericMessageImpl 
                                                 GenericTextMessage GenericTextMessageImpl)
           )
  (:use [clojure.pprint])
  (:require [clj-http.client :as client]
            [clojure.data.json :as json]))


;; Based on the Kaazing Java example:
;; http://tech.kaazing.com/documentation/jms/howto-dev-jms-java.html

(def messages-received (atom []))
(def exceptions-received (atom []))

(defn create-connection
  "Create a connection and return a connection info map describing it."
  [uri]
  (let [props (doto (Properties.)
                (.put InitialContext/INITIAL_CONTEXT_FACTORY "com.kaazing.gateway.jms.client.stomp.StompInitialContextFactory")
                (.put Context/PROVIDER_URL uri))
        ctx (InitialContext. props)
        connection-factory (.lookup ctx "ConnectionFactory")
        connection (.createConnection connection-factory nil nil)]
    (.setExceptionListener connection (proxy [ExceptionListener] []
                                        (onException [^JMSException e]
                                          (swap! exceptions-received conj
                                                 {:exception e,
                                                  :stack (with-out-str (.printStackTrace e))})
                                          (.printStackTrace e))))
    {:connection connection, :context ctx}))


(defn context-lookup
  "Look up a vaule from the connection info context."
  [connection-info k]
  {:pre [(string? k)]}
  (-> (:context connection-info)
      (.lookup k)))


(defn get-text
  "Get the text from a JMS message."
  [^javax.jms.TextMessage tm]
  (.getText tm))


(defn create-session 
  "Connect to a URI. Returns the session info map."
  [connection-info topic-name]
  (let [session (.createSession (:connection connection-info) false Session/AUTO_ACKNOWLEDGE)
        topic (context-lookup connection-info topic-name)
        consumer (.createConsumer session topic)]
    (.setMessageListener consumer (proxy [MessageListener] []
                                    (onMessage [^Message message]
                                      (try
                                        (let [text-message (cast TextMessage message)]
                                          (swap! messages-received conj {:message message, :text (get-text text-message)}))
                                        (catch JMSException e
                                          (.printStackTrace e))))))
    {:session session, :info connection-info}))


(defn start-connection
  [info]
  (.start (:connection info)))

(defn stop-connection
  [info]
  (.close (:connection info)))

(defn create-message
  "Create a text message."
  [session-info txt]
  (-> (:session session-info)
      (.createTextMessage txt)))

(defn create-producer
  "Create a message producer."
  [session-info]
  (-> (:session session-info)
      (.createProducer nil)))

(defn send-message
  "Send a message to the destination topic via the producer."
  [^MessageProducer producer ^string dest ^Message msg]
  (.send producer dest msg))

(defn test-connection [uri topic-name]
  (let [connection (create-connection uri)
        topic (context-lookup connection topic-name)
        session (create-session connection topic-name)
        producer (create-producer session)]
    (try
      (start-connection connection)
      (send-message producer topic (create-message session "Hello from JVM"))
      (Thread/sleep 5000)
      (finally
       (stop-connection connection)))))

(comment
  (do
    (reset! messages-received [])
    (reset! exceptions-received [])
    (test-connection "ws://localhost:8001/jms" "/topic/destination")
    (println "MESSAGES")
    (println @messages-received)
    (println "EXCEPTIONS")
    (println @exceptions-received))
  )