{"id":415,"date":"2013-05-29T14:52:22","date_gmt":"2013-05-29T14:52:22","guid":{"rendered":"http:\/\/putridparrot.com\/blog\/?p=415"},"modified":"2013-05-29T15:19:58","modified_gmt":"2013-05-29T15:19:58","slug":"activemq-and-nms","status":"publish","type":"post","link":"https:\/\/putridparrot.com\/blog\/activemq-and-nms\/","title":{"rendered":"ActiveMQ and NMS"},"content":{"rendered":"

Getting Started<\/strong><\/p>\n

ActiveMQ can be connected to in various ways. In this post I’m going to concentrate on NMS for .NET integration. <\/p>\n

For the sample code I’ve created a console application and using nu-get have added the packages named Apache.NMS and Apache.NMS.ActiveMQ.<\/p>\n

To begin with we need to run up ActiveMQ either locally or remotely (ensure you’ve noted down the connection string if you’re running remotely, for now I’ll assume the ActiveMQ server was started locally).<\/p>\n

The first thing our client application must do is create both the connection and session object to allow us to interact with ActiveMQ, this is done using the following code (it’s pretty self explanatory)<\/p>\n

\r\nIConnectionFactory factory = new ConnectionFactory();\r\nusing (IConnection connection = factory.CreateConnection())\r\n{\r\n   connection.Start();\r\n   using(ISession session = connection.CreateSession())\r\n   {\r\n      \/\/ do something with the session\r\n   }\r\n}\r\n<\/pre>\n
Note: both the connection and session are disposable so don;t forget to dispose of them either directly by calling Close() or indirectly using the using statement.<\/p>\n
The above code does not specify the endpoint for the connection, to handle a remote server (or one not on the default port on the local machine) we can use the following code as a replacement for the ConnectionFactory() code<\/p>\n
\r\nUri connecturi = new Uri("activemq:tcp:\/\/myserver:61616");\r\nIConnectionFactory factory = new NMSConnectionFactory(connecturi);\r\n<\/pre>\n
Topics and Queues<\/strong><\/p>\n
Best to read the ActiveMQ documentation<\/a> for a full explanation.<\/p>\n
But basically a topic works on a publish and subscribe pattern. One or more subscriptions may exist to a topic and all will receive published messages. Whereas a queue is implemented as per load balancer semantics, in that each consumer will get a message until it’s been acknowledged.<\/p>\n
Enumerating Topics and Queues<\/strong><\/p>\n
The following is pretty much taken from the source code accompanying the NMS source, I’ve removed the code for setting up the connection\/session<\/p>\n
\r\npublic const string QUEUE_ADVISORY_DESTINATION = "ActiveMQ.Advisory.Queue";\r\npublic const string TOPIC_ADVISORY_DESTINATION = "ActiveMQ.Advisory.Topic";\r\npublic const string TEMPQUEUE_ADVISORY_DESTINATION = "ActiveMQ.Advisory.TempQueue";\r\npublic const string TEMPTOPIC_ADVISORY_DESTINATION = "ActiveMQ.Advisory.TempTopic";\r\n\r\npublic const string ALLDEST_ADVISORY_DESTINATION = QUEUE_ADVISORY_DESTINATION + "," +\r\n                       TOPIC_ADVISORY_DESTINATION + "," +\r\n                       TEMPQUEUE_ADVISORY_DESTINATION + "," +\r\n                       TEMPTOPIC_ADVISORY_DESTINATION;\r\n\r\nprivate void Enumerate(string destination, Action<DestinationInfo> action)\r\n{\r\n   IDestination dest = session.GetTopic(destination);\r\n   using (IMessageConsumer consumer = session.CreateConsumer(dest))\r\n   {\r\n      IMessage advisory;\r\n\r\n      while ((advisory = consumer.Receive(TimeSpan.FromMilliseconds(2000))) != null)\r\n      {\r\n         ActiveMQMessage am = advisory as ActiveMQMessage;\r\n\t if (am != null & am.DataStructure != null)\r\n \t {\r\n\t    DestinationInfo info = am.DataStructure as DestinationInfo;\r\n \t    if (info != null)\r\n\t    {\r\n\t       action(info);\r\n\t    }\r\n\t }\r\n      }\r\n   }\t\t\t\r\n}\r\n\r\npublic void EnumerateQueues()\r\n{\r\n   Console.WriteLine("Listing all Queues on Broker:");\r\n   Enumerate(QUEUE_ADVISORY_DESTINATION, info => \r\n                Console.WriteLine("   Queue: " + info.Destination));\r\n   Console.WriteLine("Listing Complete.");\r\n}\r\n\r\npublic void EnumerateTopics()\r\n{\r\n   Console.WriteLine("Listing all Topics on Broker:");\r\n   Enumerate(TOPIC_ADVISORY_DESTINATION, info => \r\n                Console.WriteLine("   Topic: " + info.Destination));\r\n   Console.WriteLine("Listing Complete.");\r\n}\r\n\r\npublic void EnumerateDestinations()\r\n{\r\n   Console.WriteLine("Listing all Destinations on Broker:");\r\n   Enumerate(ALLDEST_ADVISORY_DESTINATION, info =>\r\n   {\r\n      string destType = info.Destination.IsTopic ? "Topic" : "Qeue";\r\n      destType = info.Destination.IsTemporary ? "Temporary" + destType : destType;\r\n      Console.WriteLine("   " + destType + ": " + info.Destination);\r\n   });\r\n   Console.WriteLine("Listing Complete.");\r\n}\r\n<\/pre>\n
I’ve also added the code for EnumerateDestinations which allows us to enumerate all both Queues and Topics.<\/p>\n
Note the use of advisory messages, see Advisory Messages<\/a><\/p>\n
Connection strings<\/strong><\/p>\n
Sample of alternate connection strings<\/p>\n
activemq:tcp:\/\/activemqhost:61616
\nstomp:tcp:\/\/activemqhost:61613<\/p>\n
Creating a producer<\/strong><\/p>\n
\r\nIDestination destination = SessionUtil.GetDestination(session, "queue:\/\/FOO.BAR");\r\nusing (IMessageConsumer consumer = session.CreateConsumer(destination))\r\n{\r\n   connection.Start();\r\n   producer.DeliveryMode = MsgDeliveryMode.Persistent;\r\n   poducer.RequestTimeout = TimeSpan.FromSeconds(10);\r\n\r\n   \/\/ create message\r\n   ITextMessage request = session.CreateTextMessage("Hello World!");\r\n   request.NMSCorrelationID = "abc";\r\n   request.Properties["myKey"] = "Cheddar";\r\n\r\n   producer.Send(request);\r\n}\r\n<\/pre>\n
Creating a consumer<\/strong><\/p>\n
\r\nIDestination destination = SessionUtil.GetDestination(session, "queue:\/\/FOO.BAR");\r\nusing (IMessageConsumer consumer = session.CreateConsumer(destination))\r\n{\r\n   connection.Start();  \r\n   consumer.Listener += msg => \r\n   {\r\n      ITextMessage message = msg as ITextMessage;\r\n      if(message != null)\r\n      {\r\n         Console.WriteLine("Message Id: " + message.NMSMessageId);\r\n         Console.WriteLine("Message text " + message.Text);\r\n      }\r\n   }\r\n}\r\n<\/pre>\n
The above consumer code works in an event driven way and thus can be used as an async consumer an alternative would be to request data from the consumer in the following manner<\/p>\n
\r\nITextMessage message = consumer.Receive() as ITextMessage;\r\nif(message != null)\r\n{\r\n   Console.WriteLine("Message Id: " + message.NMSMessageId);\r\n   Console.WriteLine("Message text " + message.Text);\r\n}\r\n<\/pre>\n
Note that ActiveMQ queues and topics can be created by simply creating a destination to use them, in the above case we’re creating a queue named FOO.BAR as in the following<\/p>\n
queue:\/\/FOO.BAR<\/p>\n
Instead of using SessionUtil.GetDestination(session, “queue:\/\/FOO.BAR”) we can also use session.GetQueue(“FOO.BAR”) and\/or session.GetTopic(“FOO.BAR”) or session.GetDestination(“queue:\/\/FOO.BAR”) and\/or session.GetDestination(“topic:\/\/FOO.BAR”).<\/p>\n
Note: if using SessionUtil.GetDestination(session, “FOO.BAR”) i.e. without the queue\/topic prefix the call defaults to a queue.<\/p>\n
Don’t forget to cleanup<\/strong><\/p>\n
As mentioned earlier, one thing to remember is that the session and connection instances need to be disposed of preferably by calling the Close() method directly when no longer required.<\/p>\n","protected":false},"excerpt":{"rendered":"
Getting Started ActiveMQ can be connected to in various ways. In this post I’m going to concentrate on NMS for .NET integration. For the sample code I’ve created a console application and using nu-get have added the packages named Apache.NMS and Apache.NMS.ActiveMQ. To begin with we need to run up ActiveMQ either locally or remotely […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"_jetpack_memberships_contains_paid_content":false,"footnotes":""},"categories":[41],"tags":[],"class_list":["post-415","post","type-post","status-publish","format-standard","hentry","category-activemq"],"jetpack_sharing_enabled":true,"jetpack_featured_media_url":"","_links":{"self":[{"href":"https:\/\/putridparrot.com\/blog\/wp-json\/wp\/v2\/posts\/415","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/putridparrot.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/putridparrot.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/putridparrot.com\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/putridparrot.com\/blog\/wp-json\/wp\/v2\/comments?post=415"}],"version-history":[{"count":9,"href":"https:\/\/putridparrot.com\/blog\/wp-json\/wp\/v2\/posts\/415\/revisions"}],"predecessor-version":[{"id":421,"href":"https:\/\/putridparrot.com\/blog\/wp-json\/wp\/v2\/posts\/415\/revisions\/421"}],"wp:attachment":[{"href":"https:\/\/putridparrot.com\/blog\/wp-json\/wp\/v2\/media?parent=415"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/putridparrot.com\/blog\/wp-json\/wp\/v2\/categories?post=415"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/putridparrot.com\/blog\/wp-json\/wp\/v2\/tags?post=415"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}