Checking Tibco EMS Queue Connected Users with Clojure

During our Tibco EMS infrastructure upgrade project, we had some rogue connections that just refused to disconnect from the old infrastructure and switch to the new Tibco EMS infrastructure. In order to track down these connections and evict those connections, I had to generate a report which would tell me which user, from which host is connected to particular queues in Tibco EMS. Here's the script that I wrote in Clojure that generates that report.

; Get destination users (and which host the user is connecting from) along with which queue/topic it is connected to.
(defn get-destination-users [server-url username password]
  (with-open [admin (TibjmsAdmin. server-url username password)]
    (let [connections  (.getConnections admin) 
          consumers    (.getConsumers admin)
          dest-userids (->> consumers
                         (map (fn [c] {:userid (.getConnectionID c) 
                                       :dest (.getDestinationName c)})))
          get-user (fn [dest-name id]
                     (let [conn (first (filter #(= id (.getID %)) connections))]
                       (if (nil? conn)
                         {:dest dest-name :user (str "Unknown user : " id) :host "unknown"}
                         {:dest dest-name :user (.getUserName conn) :host (.getHost conn)})))]
      (map (fn [x] (get-user (:dest x) (:userid x))) dest-userids))))

; Dump results in CSV format
(def results (get-destination-users server-url username password))   
(doseq [r (sort-by :user (set results))]
  (println  (str (:user r) "," (:host r) "," (:dest r))))

Working with Tibco EMS Message Topics using F# and Clojure

In my previous blog post, I showed how to connect to a Tibco EMS Queue with F# and Clojure to represent integration interoperability from .NET and Java platforms. Message queues are a way to implement the Request-Reply pattern, which is one of the many enterprise integration patterns described in the book Enterprise Integration Patterns. Another basic enterprise integration pattern is the Publish-Subscribe pattern, which can be implement via JMS topics. This blog post shows how to connect to JMS topic from .NET and Java using Tibco EMS as the JMS provider.

Here is the F# version:

#r @"C:\tibco\ems\6.3\bin\TIBCO.EMS.dll"

open System
open TIBCO.EMS

let serverUrl = "tcp://localhost:7222"
let producer = "producer"
let consumer = "consumer"
let password = "testpwd"
let topicName = "testTopic"


let subscribeToTopic serverUrl userid password topicName messageProcessor =
    async {
        let connection = (userid,password)
                         |> (new TopicConnectionFactory(serverUrl)).CreateTopicConnection
        let session = connection.CreateTopicSession(false,Session.AUTO_ACKNOWLEDGE)
        let topic = session.CreateTopic(topicName)
        let subscriber =  session.CreateSubscriber(topic)
        connection.Start()
        printf "Subscriber connected!\n"
        while true do
            try
                subscriber.Receive() |> messageProcessor
            with _ ->  ()
        connection.Close()
    }

let publishTopicMessages serverUrl  userid password topicName messages =
    let connection = (userid,password)
                     |> (new TopicConnectionFactory(serverUrl)).CreateTopicConnection
    let session = connection.CreateTopicSession(false,Session.AUTO_ACKNOWLEDGE)
    let topic = session.CreateTopic(topicName)
    let publisher = session.CreatePublisher(topic)
    connection.Start()

    messages
    |> Seq.iter (fun item -> session.CreateTextMessage(Text=item)
                             |> publisher.Send)
                             
    connection.Close()

// Just dump message to console for now
let myMessageProcessor (msg:Message) =
    msg.ToString() |> printf "%s\n"

let consumeMessageAsync = subscribeToTopic "tcp://localhost:7222" "consumer" "testpwd"


let produceMessages topicName messages = publishTopicMessages "tcp://localhost:7222" "producer" "testpwd" topicName messages 


// Asynchronously start the topic subscriber
Async.Start(consumeMessageAsync "testTopic" myMessageProcessor)


// Publish messages to the Tibco EMS topic
[ "Aslund"; "Barrayar"; "Beta Colony"; "Cetaganda"; "Escobar"; "Komarr"; "Marilac"; "Pol"; "Sergyar"; "Vervain"]
|> produceMessages "testTopic"


printf "Done!"

One thing to point out is that Tibco, unfortunately, did not implement IDisposable for it's Connection objects; perhaps in it's bid to stay faithful to the Java API. That design choice seems unfortunate to me in the sense that I no longer can leverage C#'s using keyword or F#'s use keyword to automatically close connection. I suppose it is fairly trivial to subclass the QueueConnection and TopicConnection class and add the IDisposable interface, but I feel that Tibco should have done this and developed the Tibco .NET API using idioms that are .NET specific.

Putting my rants aside, here is the equivalent Clojure code to connect to Tibco Topics:

(import '(java.util Enumeration)
        '(com.tibco.tibjms TibjmsTopicConnectionFactory)
        '(javax.jms Message JMSException  Session
                    Topic TopicConnectionFactory
                    TopicConnection TopicSession
                    TopicSubscriber))
                  
(def serverUrl "tcp://localhost:7222")
(def producer "producer")
(def consumer "consumer")
(def password "testpwd")
(def topicName "testTopic")

;------------------------------------------------------------------------------
; Subscribe to Topic asynchronously
;------------------------------------------------------------------------------
(defn subscribe-topic [server-url user password topic-name process-message]
    (future
        (with-open [connection (-> (TibjmsTopicConnectionFactory. server-url)
                                   (.createTopicConnection user password))]
            (let [session (.createTopicSession connection false Session/AUTO_ACKNOWLEDGE)
                  topic (.createTopic session  topic-name)]
                (with-open [subscriber (.createSubscriber session topic)]
                    (.start connection)
                    (loop []                       
                        (process-message (.receive subscriber))
                        (recur)))))))

;------------------------------------------------------------------------------
; Publishing to a Topic
;------------------------------------------------------------------------------
(defn publish-to-topic [server-url user password topic-name messages]
    (with-open [connection (-> (TibjmsTopicConnectionFactory. server-url)
                               (.createTopicConnection user password))]
        (let [session (.createTopicSession connection false Session/AUTO_ACKNOWLEDGE)
              topic (.createTopic session  topic-name)
              publisher (.createPublisher session topic)]
            (.start connection)
            (doseq [item messages]
                (let [message (.createTextMessage session)]
                    (.setText message item)
                    (.publish publisher message))))))
                    
                      
; Create function aliases with connection information embedded                    
(defn produce-messages [topic-name messages]
    (publish-to-topic "tcp://localhost:7222" "producer" "testpwd" topic-name messages))

(defn consume-messages [topic-name message-processor]
    (subscribe-topic "tcp://localhost:7222" "consumer" "testpwd" topic-name message-processor))

; Just dump messages to console for now
(defn my-message-processor [message]
    (println (.toString message)))
    
; Start subscribing messages asynchronously
(consume-messages "testTopic" my-message-processor)                            
    
; Publish to topic
(def my-messages '("alpha" "beta" "gamma" "delta"
                   "epsilon" "zeta" "eta" "theta"
                   "iota" "kappa" "lambda" "mu" "nu"
                   "xi", "omicron" "pi" "rho"
                   "signma" "tau" "upsilon" "phi",
                   "chi" "psi" "omega"))                    

(produce-messages  "testTopic"  my-messages)    

When I fire up both scripts, the messages published to the topic would be received by both the .NET and Java clients. With these scripts, I can easily swap out the message generators or message processors as needed for any future testing scenarios.

F#, Clojure and Message Queues on Tibco EMS

It looks like I will be getting much more hands on with Tibco EMS. Since the Tibco EMS system in use will have connections from both .NET platforms and Java platforms, I wanted to write some scripts to run some engineering tests on Tibco EMS. I decided to simulate .NET side connections with F# and Java side connections with Clojure. Taking the sample code from Tibco installation, I created the following F# script that sends messages to a queue from the sample C# code:

#r @"C:\tibco\ems\6.3\bin\TIBCO.EMS.dll"

open System
open TIBCO.EMS

let serverUrl = "tcp://localhost:7222"
let producer = "producer"
let consumer = "consumer"
let password = "testpwd"
let queueName = "testQueue"


let getQueueTextMessages serverUrl  userid password queueName messageProcessor =
    async {
        let connection = (userid,password)
                         |> (new QueueConnectionFactory(serverUrl)).CreateQueueConnection
        let session = connection.CreateQueueSession(false,Session.AUTO_ACKNOWLEDGE)
        let queue = session.CreateQueue(queueName)
        let receiver =  session.CreateReceiver(queue)
        connection.Start()
        printf "Queue connection established!"
        while true do
            try
                receiver.Receive() |> messageProcessor
            with _ ->  ()
    }


let sendQueueTextMessages serverUrl  userid password queueName messages =
    let connection = (userid,password)
                     |> (new QueueConnectionFactory(serverUrl)).CreateQueueConnection
    let session = connection.CreateQueueSession(false,Session.AUTO_ACKNOWLEDGE)
    let queue = session.CreateQueue(queueName)
    let sender = session.CreateSender(queue)
    connection.Start()

    messages
    |> Seq.iter (fun item -> session.CreateTextMessage(Text=item)
                             |> sender.Send)
                             
    connection.Close()



// Just dump message to console for now
let myMessageProcessor (msg:Message) =
    msg.ToString() |> printf "%s\n"


let consumeMessageAsync = getQueueTextMessages "tcp://localhost:7222" "consumer" "testpwd"
let produceMessages queueName messages = sendQueueTextMessages "tcp://localhost:7222" "producer" "testpwd" queueName messages 

// Start message consumer asynchronously
Async.Start(consumeMessageAsync "testQueue" myMessageProcessor)


// Send messages to the Tibco EMS   
[ "Aslund"; "Barrayar"; "Beta Colony"; "Cetaganda"; "Escobar"; "Komarr"; "Marilac"; "Pol"; "Sergyar"; "Vervain"]
|> produceMessages "testQueue"

The queue consumer is implemented asynchronously so it won't block executing subsequent statements. To test Tibco JMS from Java, here is the equivalent Clojure code:

(import '(java.util Enumeration)
        '(com.tibco.tibjms TibjmsQueueConnectionFactory)
        '(javax.jms Message JMSException  Session
                    Queue QueueBrowser 
                    QueueConnection QueueReceiver 
                    QueueSession QueueSender))
                  
(def serverUrl "tcp://localhost:7222")
(def producer "producer")
(def consumer "consumer")
(def password "testpwd")
(def queueName "testQueue")

; Consume Queue Text messages asynchronously
(defn get-queue-text-messages [server-url user password queue-name process-message]
    (future
        (with-open [connection (-> (TibjmsQueueConnectionFactory. server-url)
                                   (.createQueueConnection user password))]
            (let [session (.createQueueSession connection false Session/AUTO_ACKNOWLEDGE)
                  queue (.createQueue session  queue-name)]
                (with-open [receiver (.createReceiver session queue)]              
                    (.start connection)
                    (loop []                       
                        (process-message (.receive receiver))
                        (recur)))))))
                   
; Send multiple Text messages
(defn send-queue-text-messages [server-url user password queue-name messages]
    (with-open [connection (-> (TibjmsQueueConnectionFactory. server-url)
                               (.createQueueConnection user password))]
        (let [session (.createQueueSession connection false Session/AUTO_ACKNOWLEDGE)
              queue (.createQueue session  queue-name)
              sender (.createSender session queue)]
            (.start connection)
            (doseq [item messages]
                (let [message (.createTextMessage session)]
                    (.setText message item)
                    (.send sender message))))))


; Create function aliases with connection information embedded                    
(defn consume-messages [queue-name message-processor]
    (get-queue-text-messages  serverUrl producer password queue-name message-processor))

(defn produce-messages [queue-name messages]
    (send-queue-text-messages  serverUrl producer password queue-name messages))

; Just dump messages to console for now
(defn my-message-processor [message]
    (println (.toString message)))

    
; Start consuming messages asynchronously
(consume-messages "testQueue" my-message-processor)                            

; Send messages to queue
(def my-messages '("alpha" "beta" "gamma" "delta"
                   "epsilon" "zeta" "eta" "theta"
                   "iota" "kappa" "lambda" "mu" "nu"
                   "xi", "omicron" "pi" "rho"
                   "signma" "tau" "upsilon" "phi",
                   "chi" "psi" "omega"))                    

(produce-messages  "testQueue"  my-messages)    

With these scripts, I can easily swap in different message generators and message processors as needed for any testing purposes. When I fired up both these scripts up to the part where queue consumers are running in both F# and Clojure version and then send the messages, I could see that Tibco EMS send half the messages to my F# script and the other half to my Clojure script. Since both of these scripts run in REPL environment, I can easily adjust my level of testing as I get results.