Page History

...

3) In separate windows, start two one or more fanout listeners as follows:

...

Writing Publish/Subscribe Applications

This section describes two programs that implement Publish/Subscribe messaging using a topic exchange.

• Publisher.cS sends messages to the amq.topic exchange, using the multipart routing keys usa.news, usa.weather, europe.news, and europe.weather.
• Listener.cs creates private queues for news, weather, usa, and europe, binding them to the amq.topic exchange using bindings that match the corresponding parts of the multipart routing keys.

In this example, the publisher creates messages for topics like news, weather, and sports that happen in regions like Europe, Asia, or the United States. A given consumer may be interested in all weather messages, regardless of region, or it may be interested in news and weather for the United States, but uninterested in items for other regions. In this example, each consumer sets up its own private queues, which receive precisely the messages that particular consumer is interested in.

Running the Publish-Subscribe Examples

1) Make sure your PATH contains the directory <home>/qpid/lib

2) Make sure that a qpid broker is running:

$ ps -eaf | grep qpidd

If a broker is running, you should see the qpidd process in the output of the above
command.

3) In separate windows, start one or more topic subscribers as follows:

$ cd <home>/qpid/examples/direct

With cygwin:

$ ./example-pub-sub--Listener.exe [hostname] [portnumber]

or with mono:

$ mono ./example-pub-sub-Listener.exe [hostname] [portnumber]

You will see output similar to this:

Listening for messages ...
Declaring queue: usa
Declaring queue: europe
Declaring queue: news
Declaring queue: weather

Each topic consumer creates a set of private queues, and binds each queue to the amq.topic exchange together with a binding that indicates which messages should be routed to the queue.

4) In another window, start the topic publisher, which publishes messages to the amq.topic exchange, as follows:

$ cd <home>/qpid/examples/direct

With cygwin:

$ ./example-pub-sub-Producer.exe  [hostname] [portnumber]

or with mono:

$ mono ./example-pub-sub-Producer.exe [hostname] [portnumber]

This program has no output; the messages are routed to the message queues for each topic_consumer as specified by the bindings the consumer created.

5) Go back to the window for each topic consumer. You should see output like this:

Message: Message 0 from usa
Message: Message 0 from news
Message: Message 0 from weather
Message: Message 1 from usa
Message: Message 1 from news
Message: Message 2 from usa
Message: Message 2 from news
Message: Message 3 from usa
Message: Message 3 from news
Message: Message 4 from usa
Message: Message 4 from news
Message: Message 5 from usa
Message: Message 5 from news
Message: Message 6 from usa
Message: Message 6 from news
Message: Message 7 from usa
Message: Message 7 from news
Message: Message 8 from usa
Message: Message 8 from news
Message: Message 9 from usa
....
Message: That's all, folks! from weather
Shutting down listener for control
Message: That's all, folks! from europe
Shutting down listener for control

Now we will examine the code for each of these programs. In each section, we will discuss only
the code that must be added to the skeleton shown in Section "Creating and Closing Sessions".

Publishing Messages to a Topic Exchange

The first program in the publish/subscribe example, Publisher.cs, defines two new functions: one that publishes messages to the topic exchange, and one that indicates that no more messages are coming.

The publishMessages function publishes a series of five messages using the specified routing key.

private static void publishMessages(ClientSession session, string routing_key)
{
 IMessage message = new Message();
 // Asynchronous transfer sends messages as quickly as
 // possible without waiting for confirmation.
 for (int i = 0; i < 10; i++)
 {
    message.clearData();
    message.appendData(Encoding.UTF8.GetBytes("Message " + i));
    session.messageTransfer("amq.topic", routing_key, message);
 }
}

The noMoreMessages function signals the end of messages using the control routing key, which is reserved for control messages.

private static void noMoreMessages(ClientSession session)
{
  IMessage message = new Message();
  // And send a syncrhonous final message to indicate termination.
  message.clearData();
  message.appendData(Encoding.UTF8.GetBytes("That's all, folks!"));
  session.messageTransfer("amq.topic", "control", message);
  session.sync();
}

In the main body of the program, messages are published using four different routing keys, and then the end of messages is indicated by a message sent to a separate routing key.

publishMessages(session, "usa.news");
publishMessages(session, "usa.weather");
publishMessages(session, "europe.news");
publishMessages(session, "europe.weather");

noMoreMessages(session);

Reading Messages from the Queue

The second program in the publish/subscribe example, Listener.cs, creates a local private queue, with a unique name, for each of the four binding keys it specifies: usa.#, europe.#, #.news, and #.weather, and creates a listener.

Console.WriteLine("Listening for messages ...");
// Create a listener                    
prepareQueue("usa", "usa.#", session);
prepareQueue("europe", "europe.#", session);
prepareQueue("news", "#.news", session);
prepareQueue("weather", "#.weather", session);

The prepareQueue() method creates a queue using a queue name and a routing key supplied as arguments it then attaches a listener with the session for the created queue and subscribe for this receiving messages from the queue:

// Create a unique queue name for this consumer by concatenating
// the queue name parameter with the Session ID.     
Console.WriteLine("Declaring queue: " + queue);
session.queueDeclare(queue, Option.EXCLUSIVE, Option.AUTO_DELETE);

// Route messages to the new queue if they match the routing key.
// Also route any messages to with the "control" routing key to
// this queue so we know when it's time to stop. A publisher sends
// a message with the content "That's all, Folks!", using the
// "control" routing key, when it is finished.

session.exchangeBind(queue, "amq.topic", routing_key);
session.exchangeBind(queue, "amq.topic", "control");

// subscribe the listener to the queue
IMessageListener listener = new MessageListener(session);
session.attachMessageListener(listener, queue);
session.messageSubscribe(queue);