dotnet/Qpid.Client.Tests/interop/TopicListener.cs - qpid - Git at Google

 /*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF 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 log4net;
 using Apache.Qpid.Messaging;
 using Apache.Qpid.Client.Qms;

 namespace Apache.Qpid.Client.Tests.interop
 {
     public class TopicListener
     {
         private static ILog log = LogManager.GetLogger(typeof(TopicListener));

         /// <summary> The default AMQ connection URL to use for tests. </summary>
         const string DEFAULT_URI = "amqp://guest:guest@default/test?brokerlist='tcp://localhost:5672'";

         /// <summary> Holds the routing key for the topic to receive test messages on. </summary>
         public static string CONTROL_ROUTING_KEY = "topic_control";

         /// <summary> Holds the routing key for the queue to send reports to. </summary>
         public static string RESPONSE_ROUTING_KEY = "response";

         /// <summary> Holds the connection to listen on. </summary>
         private IConnection connection;

         /// <summary> Holds the channel for all test messages.</summary>
         private IChannel channel;

         /// <summary> Holds the producer to send report messages on. </summary>
         private IMessagePublisher publisher;

         /// <summary> Holds a flag to indicate that a timer has begun on the first message. Reset when report is sent. </summary> */
         private bool init;

         /// <summary> Holds the count of messages received by this listener. </summary> */
         private int count;

         /// <summary> Creates a topic listener using the specified broker URL. </summary>
         ///
         /// <param name="connectionUri">The broker URL to listen on.</param>
         TopicListener(string connectionUri)
         {
             log.Debug("TopicListener(string connectionUri = " + connectionUri + "): called");

             // Create a connection to the broker.
             IConnectionInfo connectionInfo = QpidConnectionInfo.FromUrl(connectionUri);
             connection = new AMQConnection(connectionInfo);

             // Establish a session on the broker.
             channel = connection.CreateChannel(false, AcknowledgeMode.AutoAcknowledge, 1);

             // Set up a queue to listen for test messages on.
             string topicQueueName = channel.GenerateUniqueName();
             channel.DeclareQueue(topicQueueName, false, true, true);

             // Set this listener up to listen for incoming messages on the test topic queue.
             channel.Bind(topicQueueName, ExchangeNameDefaults.TOPIC, CONTROL_ROUTING_KEY);
             IMessageConsumer consumer = channel.CreateConsumerBuilder(topicQueueName)
                 .Create();
             consumer.OnMessage += new MessageReceivedDelegate(OnMessage);

             // Set up this listener with a producer to send the reports on.
             publisher = channel.CreatePublisherBuilder()
                 .WithExchangeName(ExchangeNameDefaults.DIRECT)
                 .WithRoutingKey(RESPONSE_ROUTING_KEY)
                 .Create();

             connection.Start();
             Console.WriteLine("Waiting for messages...");
         }

         public static void Main(String[] argv)
         {
             // Create an instance of this listener with the command line parameters.
             new TopicListener(DEFAULT_URI);
         }

         /// <summary>
         /// Handles all message received by this listener. Test messages are counted, report messages result in a report being sent and
         /// shutdown messages result in this listener being terminated.
         /// </summary>
         ///
         /// <param name="message">The received message.</param>
         public void OnMessage(IMessage message)
         {
             log.Debug("public void onMessage(Message message = " + message + "): called");

             // Take the start time of the first message if this is the first message.
             if (!init)
             {
                 count = 0;
                 init = true;
             }

             // Check if the message is a control message telling this listener to shut down.
             if (IsShutdown(message))
             {
                 log.Debug("Got a shutdown message.");
                 Shutdown();
             }
             // Check if the message is a report request message asking this listener to respond with the message count.
             else if (IsReport(message))
             {
                 log.Debug("Got a report request message.");

                 // Send the message count report.
                 SendReport();

                 // Reset the initialization flag so that the next message is considered to be the first.
                 init = false;
             }
             // Otherwise it is an ordinary test message, so increment the message count.
             else
             {
                 count++;
             }
         }

         /// <summary> Checks a message to see if it is a shutdown control message. </summary>
         ///
         /// <param name="m">The message to check.</param>
         ///
         /// <returns><tt>true</tt> if it is a shutdown control message, <tt>false</tt> otherwise.</returns>
         private bool IsShutdown(IMessage m)
         {
             bool result = CheckTextField(m, "TYPE", "TERMINATION_REQUEST");

             //log.Debug("isShutdown = " + result);

             return result;
         }

         /// <summary> Checks a message to see if it is a report request control message. </summary>
         ///
         /// <param name="m">The message to check.</param>
         ///
         /// <returns><tt>true</tt> if it is a report request control message, <tt>false</tt> otherwise.</returns>
         private bool IsReport(IMessage m)
         {
             bool result = CheckTextField(m, "TYPE", "REPORT_REQUEST");

             //log.Debug("isReport = " + result);

             return result;
         }

         /// <summary> Checks whether or not a text field on a message has the specified value. </summary>
         ///
         /// <param name="e">The message to check.</param>
         /// <param name="e">The name of the field to check.</param>
         /// <param name="e">The expected value of the field to compare with.</param>
         ///
         /// <returns> <tt>true</tt>If the specified field has the specified value, <tt>fals</tt> otherwise. </returns>
         private static bool CheckTextField(IMessage m, string fieldName, string value)
         {
             /*log.Debug("private static boolean checkTextField(Message m = " + m + ", String fieldName = " + fieldName
                       + ", String value = " + value + "): called");*/

             string comp = m.Headers.GetString(fieldName);

             return (comp != null) && comp == value;
         }

         /// <summary> Stops the message consumer and closes the connection. </summary>
         private void Shutdown()
         {
             connection.Stop();
             channel.Dispose();
             connection.Dispose();
         }

         /// <summary> Sends the report message to the response location. </summary>
         private void SendReport()
         {
             string report = "Received " + count + ".";

             IMessage reportMessage = channel.CreateTextMessage(report);

             reportMessage.Headers.SetBoolean("BOOLEAN", false);
             //reportMessage.Headers.SetByte("BYTE", 5);
             reportMessage.Headers.SetDouble("DOUBLE", 3.141);
             reportMessage.Headers.SetFloat("FLOAT", 1.0f);
             reportMessage.Headers.SetInt("INT", 1);
             reportMessage.Headers.SetLong("LONG", 1);
             reportMessage.Headers.SetString("STRING", "hello");
             reportMessage.Headers.SetShort("SHORT", 2);

             publisher.Send(reportMessage);

             Console.WriteLine("Sent report: " + report);
         }
     }
 }
	/*
	*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF 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 log4net;
	using Apache.Qpid.Messaging;
	using Apache.Qpid.Client.Qms;

	namespace Apache.Qpid.Client.Tests.interop
	{
	public class TopicListener
	{
	private static ILog log = LogManager.GetLogger(typeof(TopicListener));

	/// <summary> The default AMQ connection URL to use for tests. </summary>
	const string DEFAULT_URI = "amqp://guest:guest@default/test?brokerlist='tcp://localhost:5672'";

	/// <summary> Holds the routing key for the topic to receive test messages on. </summary>
	public static string CONTROL_ROUTING_KEY = "topic_control";

	/// <summary> Holds the routing key for the queue to send reports to. </summary>
	public static string RESPONSE_ROUTING_KEY = "response";

	/// <summary> Holds the connection to listen on. </summary>
	private IConnection connection;

	/// <summary> Holds the channel for all test messages.</summary>
	private IChannel channel;

	/// <summary> Holds the producer to send report messages on. </summary>
	private IMessagePublisher publisher;

	/// <summary> Holds a flag to indicate that a timer has begun on the first message. Reset when report is sent. </summary> */
	private bool init;

	/// <summary> Holds the count of messages received by this listener. </summary> */
	private int count;

	/// <summary> Creates a topic listener using the specified broker URL. </summary>
	///
	/// <param name="connectionUri">The broker URL to listen on.</param>
	TopicListener(string connectionUri)
	{
	log.Debug("TopicListener(string connectionUri = " + connectionUri + "): called");

	// Create a connection to the broker.
	IConnectionInfo connectionInfo = QpidConnectionInfo.FromUrl(connectionUri);
	connection = new AMQConnection(connectionInfo);

	// Establish a session on the broker.
	channel = connection.CreateChannel(false, AcknowledgeMode.AutoAcknowledge, 1);

	// Set up a queue to listen for test messages on.
	string topicQueueName = channel.GenerateUniqueName();
	channel.DeclareQueue(topicQueueName, false, true, true);

	// Set this listener up to listen for incoming messages on the test topic queue.
	channel.Bind(topicQueueName, ExchangeNameDefaults.TOPIC, CONTROL_ROUTING_KEY);
	IMessageConsumer consumer = channel.CreateConsumerBuilder(topicQueueName)
	.Create();
	consumer.OnMessage += new MessageReceivedDelegate(OnMessage);

	// Set up this listener with a producer to send the reports on.
	publisher = channel.CreatePublisherBuilder()
	.WithExchangeName(ExchangeNameDefaults.DIRECT)
	.WithRoutingKey(RESPONSE_ROUTING_KEY)
	.Create();

	connection.Start();
	Console.WriteLine("Waiting for messages...");
	}

	public static void Main(String[] argv)
	{
	// Create an instance of this listener with the command line parameters.
	new TopicListener(DEFAULT_URI);
	}

	/// <summary>
	/// Handles all message received by this listener. Test messages are counted, report messages result in a report being sent and
	/// shutdown messages result in this listener being terminated.
	/// </summary>
	///
	/// <param name="message">The received message.</param>
	public void OnMessage(IMessage message)
	{
	log.Debug("public void onMessage(Message message = " + message + "): called");

	// Take the start time of the first message if this is the first message.
	if (!init)
	{
	count = 0;
	init = true;
	}

	// Check if the message is a control message telling this listener to shut down.
	if (IsShutdown(message))
	{
	log.Debug("Got a shutdown message.");
	Shutdown();
	}
	// Check if the message is a report request message asking this listener to respond with the message count.
	else if (IsReport(message))
	{
	log.Debug("Got a report request message.");

	// Send the message count report.
	SendReport();

	// Reset the initialization flag so that the next message is considered to be the first.
	init = false;
	}
	// Otherwise it is an ordinary test message, so increment the message count.
	else
	{
	count++;
	}
	}

	/// <summary> Checks a message to see if it is a shutdown control message. </summary>
	///
	/// <param name="m">The message to check.</param>
	///
	/// <returns><tt>true</tt> if it is a shutdown control message, <tt>false</tt> otherwise.</returns>
	private bool IsShutdown(IMessage m)
	{
	bool result = CheckTextField(m, "TYPE", "TERMINATION_REQUEST");

	//log.Debug("isShutdown = " + result);

	return result;
	}

	/// <summary> Checks a message to see if it is a report request control message. </summary>
	///
	/// <param name="m">The message to check.</param>
	///
	/// <returns><tt>true</tt> if it is a report request control message, <tt>false</tt> otherwise.</returns>
	private bool IsReport(IMessage m)
	{
	bool result = CheckTextField(m, "TYPE", "REPORT_REQUEST");

	//log.Debug("isReport = " + result);

	return result;
	}

	/// <summary> Checks whether or not a text field on a message has the specified value. </summary>
	///
	/// <param name="e">The message to check.</param>
	/// <param name="e">The name of the field to check.</param>
	/// <param name="e">The expected value of the field to compare with.</param>
	///
	/// <returns> <tt>true</tt>If the specified field has the specified value, <tt>fals</tt> otherwise. </returns>
	private static bool CheckTextField(IMessage m, string fieldName, string value)
	{
	/*log.Debug("private static boolean checkTextField(Message m = " + m + ", String fieldName = " + fieldName
	+ ", String value = " + value + "): called");*/

	string comp = m.Headers.GetString(fieldName);

	return (comp != null) && comp == value;
	}

	/// <summary> Stops the message consumer and closes the connection. </summary>
	private void Shutdown()
	{
	connection.Stop();
	channel.Dispose();
	connection.Dispose();
	}

	/// <summary> Sends the report message to the response location. </summary>
	private void SendReport()
	{
	string report = "Received " + count + ".";

	IMessage reportMessage = channel.CreateTextMessage(report);

	reportMessage.Headers.SetBoolean("BOOLEAN", false);
	//reportMessage.Headers.SetByte("BYTE", 5);
	reportMessage.Headers.SetDouble("DOUBLE", 3.141);
	reportMessage.Headers.SetFloat("FLOAT", 1.0f);
	reportMessage.Headers.SetInt("INT", 1);
	reportMessage.Headers.SetLong("LONG", 1);
	reportMessage.Headers.SetString("STRING", "hello");
	reportMessage.Headers.SetShort("SHORT", 2);

	publisher.Send(reportMessage);

	Console.WriteLine("Sent report: " + report);
	}
	}
	}