Following on from my previous post on Active MQ, I’m now going to explore creating a mechanism whereby the message can fail.
The main issue with the trial project was that it used an auto acknowledge:
using (ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge))
There are a number of considerations here; firstly, what if the message that you read errors - we want to retry; but secondly, what happens is the message repeatedly errors (this type of message is known as a poison message).
The Problem
Here’s the send code again from the last post:
string queueName = "TextQueue";
Console.WriteLine($"Adding message to queue topic: {queueName}");
string brokerUri = $"activemq:tcp://localhost:61616"; // Default port
NMSConnectionFactory factory = new NMSConnectionFactory(brokerUri);
using (IConnection connection = factory.CreateConnection())
{
connection.Start();
using (ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge))
using (IDestination dest = session.GetQueue(queueName))
using (IMessageProducer producer = session.CreateProducer(dest))
{
producer.DeliveryMode = MsgDeliveryMode.Persistent;
var msg = session.CreateTextMessage();
producer.Send(msg);
Console.WriteLine($"Sent {text} messages");
}
}
Other than splitting the message out, I haven’t changed anything. Okay, so let’s run that and check the queue:
Now, I’m going to change the receive code slightly:
string queueName = "TextQueue";
string brokerUri = $"activemq:tcp://localhost:61616"; // Default port
NMSConnectionFactory factory = new NMSConnectionFactory(brokerUri);
using (IConnection connection = factory.CreateConnection())
{
connection.Start();
using (ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge))
using (IDestination dest = session.GetQueue(queueName))
using (IMessageConsumer consumer = session.CreateConsumer(dest))
{
IMessage msg = consumer.Receive();
if (msg is ITextMessage)
{
ITextMessage txtMsg = msg as ITextMessage;
Console.WriteLine($"Received message: {txtMsg.Text}");
message = txtMsg.Text;
throw new Exception("Test"); // <-- May cause problems
return true;
}
else
{
Console.WriteLine("Unexpected message type: " + msg.GetType().Name);
}
}
}
As you can see, there is now an issue in the code; for some reason, it is repeatedly throwing an error entitled “Test”. I can’t work out why (maybe I’ll post a question on StackOverflow later), but when I run that, despite crashing, the message is read, and the queue is now empty.
Obviously, this is an issue: if that message was “DebitBankAccountWith200000” then someone is going to wish that the person that wrote this code hadn’t automatically acknowledged it.
Firstly, how do we stop the auto acknowledge?
There are basically two alternatives to auto acknowledge (there are more, but we’ll only look at two here): client acknowledge, and transactional acknowledgement. I’ll leave transactional acknowledgement for another day.
Client Acknowledge
This method is basically the manual version. You’re telling ActiveMQ that you will, or will not acknowledge the message yourself. Now, let’s alter the receive code slightly:
using (ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge))
using (IDestination dest = session.GetQueue(queueName))
using (IMessageConsumer consumer = session.CreateConsumer(dest))
{
IMessage msg = consumer.Receive();
if (msg is ITextMessage)
{
ITextMessage txtMsg = msg as ITextMessage;
Console.WriteLine($"Received message: {txtMsg.Text}");
message = txtMsg.Text;
throw new Exception("Test");
msg.Acknowledge();
return true;
}
else
{
Console.WriteLine("Unexpected message type: " + msg.GetType().Name);
}
}
As you can see, I’ve changed two main parts here; the first is that I’ve changed that AcknowledgmentMode to ClientAcknowledge and I’ve added a call to the acknowledge method on the message.
Now let’s re-run the send and receive and see what happens to the queue.
Unfortunately, I still haven’t worked out why it’s crashing, but here’s the queue; still safely with the message:
We had an error, it crashed, but because it was never acknowledged, it’s still safe and sound in the queue. When we run the receive again, hopefully the bug will have magically disappeared and the message will successfully process.
Poison Messages
The concept of a poison message is where the issue with the message, resides in the message; the situation described above is not a poison message because the message is fine; but code is erroring. Once the code above is fixed, the message can be processed; however, let’s have a look at a different error scenario; here’s some new receive code:
string queueName = "TextQueue";
string brokerUri = $"activemq:tcp://localhost:61616"; // Default port
NMSConnectionFactory factory = new NMSConnectionFactory(brokerUri);
using (IConnection connection = factory.CreateConnection())
{
connection.Start();
using (ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge))
using (IDestination dest = session.GetQueue(queueName))
using (IMessageConsumer consumer = session.CreateConsumer(dest))
{
IMessage msg = consumer.Receive();
if (msg is ITextMessage)
{
ITextMessage txtMsg = msg as ITextMessage;
Console.WriteLine($"Received message: {txtMsg.Text}");
message = txtMsg.Text;
if (message.First() != 't')
throw new Exception("Message is invalid");
msg.Acknowledge();
return true;
}
else
{
Console.WriteLine("Unexpected message type: " + msg.GetType().Name);
}
}
}
This time, we have some code that actually processes the message and, based on the contents, does something; in this case, it throws an error where the message doesn’t start with ‘t’. So, the rules are simple; messages start with ‘t’. Let’s run the send code again and try some messages:
And now let’s receive these messages (incidentally, while testing this, my notes on starting two projects might be useful):
Okay - so we’ve come across a message that we can’t process. This has yet to be acknowledged, so it’s still safe and sound in the queue. We’ll simply restart the listener and pick it up:
Ah - okay. So, we have a problem. “nexttest” is causing an error with the queue, but if we don’t acknowledge it, we’re going to keep picking it up and erroring.
The Antidote
Once we know that the message is causing a problem, we can send it to a special queue; here’s the code to capture the error:
ITextMessage txtMsg = msg as ITextMessage;
try
{
Console.WriteLine($"Received message: {txtMsg.Text}");
message = txtMsg.Text;
if (message != null && message.First() != 't')
{
// The message has a problem, and so we need to file it away without losing it
throw new Exception("Message is invalid");
}
msg.Acknowledge();
return true;
}
catch
{
ResendMsg(session, msg);
msg.Acknowledge(); // Acknoweledge the message from the original queue
}
And here’s the new method, ResendMsg:
private static void ResendMsg(ISession session, IMessage msg)
{
var deadLetterQueue = new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("ActiveMQ.DLQ");
IMessageProducer producer = session.CreateProducer(deadLetterQueue);
producer.Send(msg);
}
The first time this executes, it will throw and catch the error, and then resend to a dead letter queue:
Subsequent runs can proceed past the problem message, and the message itself remains intact:
