Objectives
This sample demonstrates how durable or non-durable topics can be created and used in WSO2 Message Broker using the RabbitMQ .NetNET/C# client. It first introduces a sample .Net client by the name "TopicPublisher" NET named TopicPublisher, which is used to publish messages to a known, created topic in WSO2 Message Broker, and then introduces a sample .Net client by the name "TopicConsumer" NET client named TopicConsumer to listen on for messages and print message content in contents to the console.
Prerequisites
In order to To run this code sample, you need to must download and add the RabbitMQ.Client.dll file as a reference in your .net NET project. You can download that dll file from this website this file from http://www.rabbitmq.com/dotnet.html or here the WSO2 repository.
Running the
...
sample
Prior to Before running following "the TopicPublisher " classwe need to class, you must register at least one " one TopicConsumer " binding prior before sending messages to the topic . This can be done by either,
...
by doing one of the following:
- Log into WSO2 Message Broker management console and create a topic named 'test-topic' (
...
- from the Main menu, choose Topics -> Add
...
- ).
OR
...
- Run the TopicConsumer
...
- class depicted below
...
- , which will register a binding to that topic. When you have run the
...
- TopicConsumer class, you will see the topic subscription it created
...
- is visible in the management console
...
- when you choose Topics -> Browse
...
- .
...
You can now use the following " TopicPublisher " .Net NET client , to send messages can be sent to the 'test-topic' topic created earlier.
| Code Block | ||
|---|---|---|
| ||
/*
* Copyright (c) 2005-2010, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
* WSO2 Inc. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
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_8_QPID;
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");
// 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", null, Encoding.UTF8.GetBytes(message));
}
}
}
}
} |
Next, execute the following " TopicConsumer " .Net NET client, using which receives messages can be received from 'test-topic'.
| Code Block | ||
|---|---|---|
| ||
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_8_QPID;
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(message);
ch.BasicAck(e.DeliveryTag, false);
}
catch (OperationCanceledException e)
{
Console.WriteLine(e);
break;
}
}
}
}
}
}
} |
...