Tag Archives: ActiveMQ

A C# Programmer’s Guide to Queues and Sending a Message with Azure Service Bus

I have previously written about message queue systems. The big two, as far as I can see, are Active MQ and RabbitMQ.

Microsoft have always had MSMQ*, but it’s not really a message broker as such (I believe that you can get similar behaviour using NServiceBus, but have never tried that myself). However, with Azure comes the Azure Service Bus.

The first thing that you need to do is set-up an Azure account. Note that Microsoft offer Azure as a paid service, and so this is not free. However, they also offer free trials and free Azure credit if you have an MSDN.

Log on to:

https://portal.azure.com

Namespace

Namespaces are an important concept in Azure. They basically allow you to split a single Azure account across many functions, but what that means is that everything you do relates to a specific namespace.

To add one, first, pick a pricing tier:

Make sure that your Namepsace isn’t taken:

You’ll then get an alert to say it worked:

If you refresh, you should now see your namespace:

Create Test Project

I always try to start with a console app when trying new stuf. Add NuGet reference:

It is my understanding that, as with ActiveMQ and RabbitMQ, these client libraries are an abstraction over a set of HTTP Post calls. In the case of Azure, I believe that, behind the scenes, it uses WCF to handle all this.

Using the Namespace

Using a message queue system such as RabbitMQ or ActiveMQ, you need a message queue server, and a URL that relates to it. However, one of the things Azure allows you to do is to abstract that; for example:

        static void Main(string[] args)
        {
            Console.WriteLine($"Getting service bus URI...");
            Uri uri = ServiceBusEnvironment.CreateAccessControlUri("pcm-servicebustest");
            Console.WriteLine($"Service Bus URI: {uri.ToString()}");
            Console.ReadLine();
        }

Tells me what the URI of the message queue broker is:

Adding a message to a queue

In order to do anything with a message queue in Azure, you need a token; effectively, this provides a level of security

Tokens

Get the key:

You can store these details in the app/web.config, or you can use them programmatically:

        private static TokenProvider GetTokenProvider(Uri uri)
        {
            Console.WriteLine($"Getting token...");
            TokenProvider tp = TokenProvider.CreateSharedAccessSignatureTokenProvider("RootManageSharedAccessKey", "JWh82nkstIAi4w5tW6MEj7GKQfoiZlwBYjHx9wfDqdA=");                                                

            Console.WriteLine($"Token {tp.ToString()}");
            return tp;
        }

Queues

Putting the above calls together, we can now create a queue in Azure:

        private static void CreateNewQueue(Uri uri, TokenProvider tokenProvider)
        {
            Console.WriteLine($"Creating new queue...");
            NamespaceManager nm = new NamespaceManager(uri, tokenProvider);

            Console.WriteLine($"Created namespace manager for {nm.Address}");
            if (nm.QueueExists("TestQueue"))
            {
                Console.WriteLine("Queue already exists");
            }
            else
            {
                Console.WriteLine("Creating new queue");
                QueueDescription qd = nm.CreateQueue("TestQueue");
            }
        }

Incidentally, the act of creating a queue appears to have cost £0.24 GBP. If you have MSDN, you should get £40 GBP credit each month (at the time of writing).

Now we have a queue, let’s put some messages on it.

Adding a message

        private static void AddNewMessage(string id, string messageBody, string queueName)
        {
            BrokeredMessage message = new BrokeredMessage(messageBody)
            {
                MessageId = id
            };

            string connectionString = GetConnectionString();
            
            QueueClient queueClient = QueueClient.CreateFromConnectionString(connectionString, queueName);
            queueClient.Send(message);
        }

The Connection String can be found here:

We can now see that a message has, indeed, been added to the queue:

At this time, this is about as much as you can see from this portal.

Errors

These are some errors that I encountered during the creation of this post, and their solutions.

System.UnauthorizedAccessException

System.UnauthorizedAccessException: ‘The token provider was unable to provide a security token while accessing ‘https://pcm-servicebustest-sb.accesscontrol.windows.net/WRAPv0.9/’. Token provider returned message: ‘The remote name could not be resolved: ‘pcm-servicebustest-sb.accesscontrol.windows.net”.’

The cause is not an invalid secret

That’s because this line:

TokenProvider tp = TokenProvider.CreateSharedSecretTokenProvider("RootManageSharedAccessKey", "jjdsjdsjk");

Gives the error:

System.ArgumentException: ‘The ‘issuerSecret’ is invalid.’

The fix…

This code is littered throughout the web:

TokenProvider tp = TokenProvider.CreateSharedSecretTokenProvider("RootManageSharedAccessKey", "jjdsjdsjk");

But the correct code was:

TokenProvider tp = TokenProvider.CreateSharedAccessSignatureTokenProvider("RootManageSharedAccessKey", "JWh82nkstIAi4w5tW6MEj7GKQfoiZlwBYjHx9wfDqdA=");                                                

System.ArgumentNullException: ‘Queue name should be specified as EntityPath in connectionString.’

Or: 40400: Endpoint not found.

Microsoft.ServiceBus.Messaging.MessagingEntityNotFoundException: ‘40400: Endpoint not found., Resource:sb://pcm-servicebustest.servicebus.windows.net/atestqueue. TrackingId:48de75d7-fb01-4fa9-b72e-20a5dc090a8d_G11, SystemTracker:pcm-servicebustest.servicebus.windows.net:aTestQueue, Timestamp:5/25/2017 5:23:27 PM

Means (obviously) that the following code:

QueueClient.CreateFromConnectionString(connectionString, queueName);

Either doesn’t have the queue name, or it is wrong.

References

https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-exceptions

https://blogs.msdn.microsoft.com/brunoterkaly/2014/08/07/learn-how-to-create-a-queue-place-and-read-a-message-using-azure-service-bus-queues-in-5-minutes/

https://stackoverflow.com/questions/18558299/servicebus-throws-401-unauthorized-error

https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-queues-topics-subscriptions

https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dotnet-how-to-use-topics-subscriptions

https://msdn.microsoft.com/en-us/library/jj542433.aspx?f=255&MSPPError=-2147217396

https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dotnet-multi-tier-app-using-service-bus-queues

https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dotnet-get-started-with-queues

* Microsoft probably haven’t ALWAYS had MSMQ. There was probably a time in the early 90’s where they didn’t have a message queue system at all.

A C# Programmer’s Guide to Installing, Running and Messaging with ActiveMQ

I’ve recently been experimenting with message queues. I’ve used MSMQ in the past, but never with any complexity, and so I thought I’d spend some time investigating ActiveMQ. There are a number of articles and courses out there, but for some reason, C# seems to be the poor relation. So, here’s a C# programmer’s guide to installing and running ActiveMQ.

Download and Run ActiveMQ

Active MQ is from Apache, and the download link is here.

The specific version that I was working with was 5.14.0.

Once you’ve downloaded the archive, extract it; I extracted it to my Downloads folder, so the next step is to navigate to that directory in a command prompt:

apache-activemq-5.14.0-bin\apache-activemq-5.14.0\bin

Then type:

activemq start

There is a sample application that comes in the box, and it can be found here:

apache-activemq-5.14.0-bin\apache-activemq-5.14.0\examples\openwire\csharp\ActiveMQExamples

The examples that they supply do work out of the box, and they are not a bad place to start, if you don’t want to read the rest of this post.

Queues versus Topics

It is important for reasons that will become apparent shortly, to understand how and why these two concepts differ before writing any code. Let’s start with Topics. These are effectively a way to communicate between two end points; the important thing here is that there must be both for it to work. When you publish a topic message, it is published to any “listeners”. If your app wasn’t listening then that’s hard luck. The use cases here are situations whereby a message might be time sensitive; for example, a stock price had just changed or a server needs the client to refresh because there is more data. There are such things as durable topics, but for now, let’s leave topics as described here.

Queues on the other hand have a persistent nature. Once you add a message to the queue, it will remain there until it is handled. Use cases for this might include a notification to send an e-mail, a chat program, or a request to place a sales order. The queue will be read on a first in, first out basis, and so you can load balance a queue: that is, you can have n listeners, and they will all process the messages in order from the queue. If you were to do this with the topic, they would all receive the same message at the same time.

Publish and Subscribe to a Topic

Start off by creating a new console application: you might want to call it something like SendMessage, or Blancmange. Then, add the ActiveMQ NuGet package.

Here’s the code:

static void Main(string[] args)
{
    while (true)
    {
        string text = Console.ReadLine();
        if (string.IsNullOrWhiteSpace(text)) return;
        SendNewMessage(text);
    }
}

private static void SendNewMessage(string text)
{ 
    string topic = "TextQueue";

    Console.WriteLine($"Adding message to queue topic: {topic}");

    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.GetTopic(topic))
        using (IMessageProducer producer = session.CreateProducer(dest))
        { 
            producer.DeliveryMode = MsgDeliveryMode.NonPersistent;

            producer.Send(session.CreateTextMessage(text));
            Console.WriteLine($"Sent {text} messages");
        }
    }       		      
}

The important thing here is the broker address. When you set-up ActiveMQ, 61616 is the default port, but obviously this, along with the various security settings, etc, can be changed.

This results in:

activemq1

So, it looks like we’re sending a message. We can check whether or not we are by navigating to:

http://localhost:8161/

This provides you with an admin site. The default username/password is: admin/admin

Then navigate to the topics:

http://localhost:8161/admin/topics.jsp

And you should see the messages queued up:

activemq2

As you can see, TextQueue has 4 messages at the minute.

Subscriber

Now we need to write a subscriber for those messages.

static void Main(string[] args)
{
    Console.WriteLine("Waiting for messages");
 
    // Read all messages off the queue
    while (ReadNextMessage())
    {
        Console.WriteLine("Successfully read message");
    }
 
    Console.WriteLine("Finished");
}
 
static bool ReadNextMessage()
{            
    string topic = "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.GetTopic(topic))
        using (IMessageConsumer consumer = session.CreateConsumer(dest))
        {
            IMessage msg = consumer.Receive();
            if (msg is ITextMessage)
            {
                ITextMessage txtMsg = msg as ITextMessage;
                string body = txtMsg.Text;
 
                Console.WriteLine($"Received message: {txtMsg.Text}");
 
                return true;
            }
            else
            {
                Console.WriteLine("Unexpected message type: " + msg.GetType().Name);
            }
        }                
    }
 
    return false;            
}

The eagle eyed amongst you might notice that the code is almost identical; you need the same NuGet package for both the publisher and subscriber.

Topics Caveat

Okay, there’s a hugely important caveat here, which a smarter man than me would have instantly realised: if you run the subscriber now, nothing will happen. This is because the topic messages are only sent to active subscribers. In order for the above code to work, the subscriber needs to be running when the messages are sent.

So. Providing that you have an active subscriber when you publish your message, the above code will send whatever you type into the console to the subscriber.

Queues

So, the code for using queues looks very similar, but is conceptually different. Here’s the SendMessage code:

static void Main(string[] args)
{
    while (true)
    {
        string text = Console.ReadLine();
        if (string.IsNullOrWhiteSpace(text)) return;
        SendNewMessageQueue(text);
    }
}

private static void SendNewMessageQueue(string text)
{ 
    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.NonPersistent;

            producer.Send(session.CreateTextMessage(text));
            Console.WriteLine($"Sent {text} messages");
        }
    }            
}
 

This can be run and, again, you can check the queue:

activemq3

But, crucially, the following code; even if run afterwards, will still read the queue:

static void Main(string[] args)
{
    Console.WriteLine("Waiting for messages");
 
    // Read all messages off the queue
    while (ReadNextMessageQueue())
    {
        Console.WriteLine("Successfully read message");
    }
 
    Console.WriteLine("Finished");
}
 
static bool ReadNextMessageQueue()
{            
    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;
                string body = txtMsg.Text;
 
                Console.WriteLine($"Received message: {txtMsg.Text}");
 
                return true;
            }
            else
            {
                Console.WriteLine("Unexpected message type: " + msg.GetType().Name);
            }
        }                
    }
 
    return false;            
}

Here it is:

activemq4

Conclusion

So, it’s quite straightforward to get something working out of the box with ActiveMQ – certainly easier than I expected.