You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@activemq.apache.org by Todd <to...@gmail.com> on 2011/09/13 00:10:40 UTC
NMS Consumer Not Receiving Messages
I've poked around a bit and haven't found anything in the forum or docs.
I created what is supposed to be an asynchronous consumer. I found some code
in the NMS docs that I adapted. See below. (You can ignore the XML
deserialization stuff.)
This worked once or twice, but I cannot get it to work anymore. What might
be wrong with this code?
Thanks for your help.
-Todd
using System;
using Apache.NMS;
using Apache.NMS.Util;
using model;
using System.Xml.Serialization;
using System.IO;
using Domain.util;
namespace Service.AuditTrail
{
class AuditTrailService
{
private IConnectionFactory factory;
private string receiveQueueName;
IConnection connection;
ISession session;
IDestination destination;
IMessageConsumer consumer;
private AuditTrailService(string hostUri, string receiveQueue)
{
Uri uri = new Uri(hostUri);
factory = new NMSConnectionFactory(uri);
receiveQueueName = receiveQueue;
}
private void Start()
{
Console.WriteLine("Starting AuditTrailService");
SetupQueueListener();
}
#region Helper Methods
private void SetupQueueListener()
{
using (IConnection connection = factory.CreateConnection())
using (ISession session = connection.CreateSession())
{
IDestination destination =
SessionUtil.GetDestination(session, receiveQueueName);
using (IMessageConsumer consumer =
session.CreateConsumer(destination))
{
connection.Start();
consumer.Listener += new
MessageListener(OnMessageReceived);
}
}
}
private void OnMessageReceived(IMessage receivedMessage)
{
Console.WriteLine("Audit Trail Service message received...");
ITextMessage textMessage = receivedMessage as ITextMessage;
if (textMessage != null)
{
message message = ParseMessageText(textMessage.Text);
AuditRecordManager manager = new AuditRecordManager();
manager.addAuditRecord(message);
}
else
{
Console.Error.WriteLine("The received message was not an
ITextMessage.");
}
}
private static message ParseMessageText(string text)
{
XmlSerializer ser = new XmlSerializer(typeof(message));
StringReader reader = new StringReader(text);
return (message)ser.Deserialize(reader);
}
#endregion
#region Main Method
static void Main(string[] args)
{
AuditTrailService service = new
AuditTrailService("activemq:tcp://localhost:61616",
"queue://from.srv.01.to.srv.07");
service.Start();
Console.WriteLine("Press enter to terminate service.");
Console.Read();
}
#endregion
}
}
--
View this message in context: http://activemq.2283324.n4.nabble.com/NMS-Consumer-Not-Receiving-Messages-tp3808779p3808779.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
Re: NMS Consumer Not Receiving Messages
Posted by Timothy Bish <ta...@gmail.com>.
On Mon, 2011-09-12 at 15:10 -0700, Todd wrote:
> I've poked around a bit and haven't found anything in the forum or docs.
>
> I created what is supposed to be an asynchronous consumer. I found some code
> in the NMS docs that I adapted. See below. (You can ignore the XML
> deserialization stuff.)
>
> This worked once or twice, but I cannot get it to work anymore. What might
> be wrong with this code?
>
> Thanks for your help.
>
> -Todd
>
> using System;
> using Apache.NMS;
> using Apache.NMS.Util;
> using model;
> using System.Xml.Serialization;
> using System.IO;
> using Domain.util;
>
> namespace Service.AuditTrail
> {
> class AuditTrailService
> {
> private IConnectionFactory factory;
> private string receiveQueueName;
> IConnection connection;
> ISession session;
> IDestination destination;
> IMessageConsumer consumer;
>
>
> private AuditTrailService(string hostUri, string receiveQueue)
> {
> Uri uri = new Uri(hostUri);
> factory = new NMSConnectionFactory(uri);
> receiveQueueName = receiveQueue;
> }
>
> private void Start()
> {
> Console.WriteLine("Starting AuditTrailService");
> SetupQueueListener();
> }
>
> #region Helper Methods
> private void SetupQueueListener()
> {
> using (IConnection connection = factory.CreateConnection())
> using (ISession session = connection.CreateSession())
> {
> IDestination destination =
> SessionUtil.GetDestination(session, receiveQueueName);
>
> using (IMessageConsumer consumer =
> session.CreateConsumer(destination))
> {
> connection.Start();
> consumer.Listener += new
> MessageListener(OnMessageReceived);
> }
> }
> }
>
Look closely at the above method. It creates the resources in a using
block, as well as creating them as temporaries in the method and not
storing them as member data of you class. The using block will call
dispose once it completes, and the object are available for GC so they
won't stick around anyway.
> private void OnMessageReceived(IMessage receivedMessage)
> {
> Console.WriteLine("Audit Trail Service message received...");
> ITextMessage textMessage = receivedMessage as ITextMessage;
>
> if (textMessage != null)
> {
> message message = ParseMessageText(textMessage.Text);
> AuditRecordManager manager = new AuditRecordManager();
>
> manager.addAuditRecord(message);
> }
> else
> {
> Console.Error.WriteLine("The received message was not an
> ITextMessage.");
> }
> }
>
> private static message ParseMessageText(string text)
> {
> XmlSerializer ser = new XmlSerializer(typeof(message));
> StringReader reader = new StringReader(text);
> return (message)ser.Deserialize(reader);
> }
> #endregion
>
> #region Main Method
> static void Main(string[] args)
> {
> AuditTrailService service = new
> AuditTrailService("activemq:tcp://localhost:61616",
>
> "queue://from.srv.01.to.srv.07");
> service.Start();
>
> Console.WriteLine("Press enter to terminate service.");
> Console.Read();
> }
> #endregion
> }
> }
>
>
> --
> View this message in context: http://activemq.2283324.n4.nabble.com/NMS-Consumer-Not-Receiving-Messages-tp3808779p3808779.html
> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
--
Tim Bish
------------
FuseSource
Email: tim.bish@fusesource.com
Web: http://fusesource.com
Twitter: tabish121
Blog: http://timbish.blogspot.com/