This documentation is for WSO2 Message Broker 3.0.0. View documentation for the latest release.
Due to a known issue do not use JDK1.8.0_151 with WSO2 products. Use JDK 1.8.0_144 until JDK 1.8.0_162-ea is released.
Skip to end of metadata
Go to start of metadata

This sample demonstrates how durable or non-durable topics can be created and used in WSO2 Message Broker using the RabbitMQ .NET/C# client. It first introduces a sample .NET client named TopicPublisher, that publishes messages to a known, created topic in Message Broker. Then it introduces a sample .NET client named TopicConsumer that listens for messages and prints message contents to the console.

Prerequisites

To run this sample:

Building the sample

  1. Create a TopicConsumer .NET client to receive messages from the test-topic topic by adding a class with the following code.

    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using RabbitMQ.Client;
     
    namespace MB_TopicClient
    {
        class TopicConsumer
        {
            static void Main(string[] args)
            {
                TopicConsumer topicConsumer = new TopicConsumer();
                topicConsumer.GetMessages();
            }
     
            public void GetMessages()
            {
                //Setup the connection with the message broker
                ConnectionFactory factory = new ConnectionFactory();
                IProtocol protocol = Protocols.AMQP_0_9_1;
                factory.VirtualHost = "/carbon";
                factory.UserName = "admin";
                factory.Password = "admin";
                factory.HostName = "localhost";
                factory.Port = 5672;
                factory.Protocol = protocol;
                using (IConnection conn = factory.CreateConnection())
                {
                    using (IModel ch = conn.CreateModel())
                    {
                        // Declare a topic exchange to be bound to retrieve messages, here we have used the default topic exchange of WSO2 MB
                        ch.ExchangeDeclare("amq.topic", "topic");
                        // Declare a topic name, here we use a non-durable topic. To make it durable use the 2nd parameter as 'true'  
                        ch.QueueDeclare("test-topic", false, false, false, null);
                        // Bind the Topic in to the exchange
                        ch.QueueBind("test-topic", "amq.topic", "test-topic");
                        // Declare a consumer which listens on the messages published to 'test-topic' topic, we need to declare an exclusive subscriber, in order to get this work.
                        // The syntax is BasicConsume(<queuename>, <noAck>,<consumerTag>, <noLocal>, <exclusive>, <arguments>, <Consumer>)
                        QueueingBasicConsumer consumer = new QueueingBasicConsumer(ch);
                        ch.BasicConsume("test-topic", false, "1", false, true, null, consumer);
                        while (true)
                        {
                            try
                            {
                                RabbitMQ.Client.Events.BasicDeliverEventArgs e = (RabbitMQ.Client.Events.BasicDeliverEventArgs)consumer.Queue.Dequeue();
                                byte[] body = e.Body;
                                string message = Encoding.UTF8.GetString(body);
                                Console.WriteLine("Received Message : " + message);
                                ch.BasicAck(e.DeliveryTag, false);
                            }
                            catch (OperationCanceledException e)
                            {
                                Console.WriteLine(e);
                                break;
                            }
                        }
                    }
                }
            }
        }
    }

    At least one TopicConsumer binding should exist before sending messages to the topic. Therefore, this TopicConsumer class should be run before the TopicPublisher class. Alternatively, you can manually create the test-topic topic in the MB Management Console. See Adding Topics for detailed instructions.

  2. Create a TopicPublisher .NET client to send messages to the test-topic topic by adding a class with the following code.

    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using RabbitMQ.Client;
     
    namespace MB_Topic_Publisher
    {
        class TopicPublisher
        {
            static void Main(string[] args)
            {
                TopicPublisher topicPublisher = new TopicPublisher();
                topicPublisher.PublishMessage("Test Message");
                Console.WriteLine("Message Sent..");
                Console.ReadLine();
            }
     
            public void PublishMessage(string message)
            {
                //Setup the connection with the message broker
                ConnectionFactory factory = new ConnectionFactory();
                IProtocol protocol = Protocols.AMQP_0_9_1;
                factory.VirtualHost = "/carbon";
                factory.UserName = "admin";
                factory.Password = "admin";
                factory.HostName = "localhost";
                factory.Port = 5672;
                factory.Protocol = protocol;
                using (IConnection conn = factory.CreateConnection())
                {
                    using (IModel ch = conn.CreateModel())
                    {
                        // Declare a topic exchange to publish messages, here we have used the default topic exchange of WSO2 MB
                        ch.ExchangeDeclare("amq.topic", "topic");
                        IBasicProperties basicProperties = ch.CreateBasicProperties();
                        //Setting JMS Message ID.
                        basicProperties.MessageId = "ID:" + System.Guid.NewGuid().ToString();
                        //Setting content-type for message as we are sending a text message.
                        basicProperties.ContentType = "text/plain";
                        // Publish the message to the exchange, it will send it to the routing key which is our name 'myTopic'. 
                        // The syntax is ch.BasicPublish(<exchange_name>, <topic_name>, <message_properties>,<message_body>)
                        ch.BasicPublish("amq.topic", "test-topic", basicProperties, Encoding.UTF8.GetBytes(message));
                    }
                }
            }
        }
    }
  3. Add a Main.java class defining the method to call both the classes mentioned above.

Executing the sample

Run this sample from your C# project.

  • No labels