RC9/qpid/dotnet/Qpid.Integration.Tests/interactive/FailoverTest.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 System.Runtime.InteropServices;
 using System.Threading;
 using log4net;
 using NUnit.Framework;
 using Apache.Qpid.Client.Qms;
 using Apache.Qpid.Client;
 using Apache.Qpid.Messaging;

 namespace Apache.Qpid.Integration.Tests.interactive
 {
     [TestFixture, Category("Interactive")]
     public class FailoverTest : IConnectionListener
     {
         private static readonly ILog _log = LogManager.GetLogger(typeof(FailoverTest));

         /// <summary>Specifies the number of times to run the test cycle.</summary>
         const int NUM_MESSAGES = 10;

         /// <summary>Determines how many messages to send within each commit.</summary>
         const int COMMIT_BATCH_SIZE = 1;

         /// <summary>Specifies the duration of the pause to place between each message sent in the test.</summary>
         //const int SLEEP_MILLIS = 1;

         /// <summary>Specified the maximum time in milliseconds to wait for the test to complete.</summary>
         const int TIMEOUT = 10000;

         /// <summary>Defines the number of test messages to send, before prompting the user to fail a broker.</summary>
         const int FAIL_POINT = 5;

         /// <summary>Specified the ack mode to use for the test.</summary>
         AcknowledgeMode _acknowledgeMode = AcknowledgeMode.AutoAcknowledge;

         /// <summary>Determines whether this test runs transactionally or not. </summary>
         bool transacted = false;

         /// <summary>Holds the connection to run the test over.</summary>
         AMQConnection _connection;

         /// <summary>Holds the channel for the test message publisher. </summary>
         IChannel publishingChannel;

         /// <summary>Holds the test message publisher. </summary>
         IMessagePublisher publisher;

         /// <summary>Used to keep count of the number of messages sent. </summary>
         int messagesSent;

         /// <summary>Used to keep count of the number of messages received. </summary>
         int messagesReceived;

         /// <summary>Used to wait for test completion on. </summary>
         private static object testComplete = new Object();

         /// <summary>Used to wait for failover completion on. </summary>
 	private static object failoverComplete = new Object();

         bool failedOver=false;

         /// <summary>Used to record the extra message count (1) if the message sent right after failover actually made it to the new broker.</summary>
         int _extraMessage = 0;

         /// <summary>
         /// Creates the test connection with a fail-over set up, and a producer/consumer pair on that connection.
         /// </summary>
         /// [SetUp]
         public void Init(IConnectionInfo connectionInfo)
         {
 	    //log4net.Config.BasicConfigurator.Configure();
             // Reset all counts.
             messagesSent = 0;
             messagesReceived = 0;
             failedOver=false;
             _extraMessage = 0;

 	    PromptAndWait("Ensure both brokers are running, then press Enter");

             // Create a connection for the test.
             _connection = new AMQConnection(connectionInfo);
             _connection.ConnectionListener = this;

             // Create a consumer to receive the test messages.
             IChannel receivingChannel = _connection.CreateChannel(false, _acknowledgeMode);

             string queueName = receivingChannel.GenerateUniqueName();
             receivingChannel.DeclareQueue(queueName, false, true, true);
             receivingChannel.Bind(queueName, "amq.direct", queueName);

             IMessageConsumer consumer = receivingChannel.CreateConsumerBuilder(queueName)
                 .WithPrefetchLow(30)
                 .WithPrefetchHigh(60).Create();

             consumer.OnMessage = new MessageReceivedDelegate(OnMessage);
             _connection.Start();

             // Create a publisher to send the test messages.
             publishingChannel = _connection.CreateChannel(transacted, AcknowledgeMode.NoAcknowledge);
             publisher = publishingChannel.CreatePublisherBuilder()
                 .WithRoutingKey(queueName)
                 .Create();

             _log.Debug("connection = " + _connection);
             _log.Debug("connectionInfo = " + connectionInfo);
             _log.Debug("connection.AsUrl = " + _connection.toURL());
             _log.Debug("AcknowledgeMode is " + _acknowledgeMode);
         }

         /// <summary>
         /// Clean up the test connection.
         /// </summary>
         [TearDown]
         public virtual void Shutdown()
         {
  	    if (!failedOver)
 	    {
                  Assert.Fail("The failover callback never occured.");
             }

             Console.WriteLine("Test done shutting down");
 	    Thread.Sleep(2000);
             _connection.Close();
         }

         /// <summary>
         /// Runs a failover test, building up the connection information from its component parts. In particular the brokers
         /// to fail between are seperately added into the connection info.
         /// </summary>
         /*[Test]
         public void TestWithBasicInfo()
         {
             _log.Debug("public void TestWithBasicInfo(): called");

             // Manually create the connection parameters.
             QpidConnectionInfo connectionInfo = new QpidConnectionInfo();
             connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 5672, false));
             connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 5673, false));

             Init(connectionInfo);
             DoFailoverTest();
         }*/

         /// <summary>
         /// Runs a failover test, with the failover configuration specified in the Qpid connection URL format.
         /// </summary>
         [Test]
         public void TestWithUrl()
         {
             _log.Debug("public void runTestWithUrl(): called");

             // Parse the connection parameters from a URL.
             String clientId = "failover" + DateTime.Now.Ticks;
             string defaultUrl = "amqp://guest:guest@" + clientId + "/test" +
                 "?brokerlist='tcp://localhost:9672;tcp://localhost:9673'&failover='roundrobin'";
             IConnectionInfo connectionInfo = QpidConnectionInfo.FromUrl(defaultUrl);

             Init(connectionInfo);
             DoFailoverTest(0);
         }

         /// <summary>
         /// Send the test messages, prompting at the fail point for the user to cause a broker failure. The test checks that all messages sent
         /// are received within the test time limit.
         /// </summary>
         ///
         /// <param name="connectionInfo">The connection parameters, specifying the brokers to fail between.</param>
         void DoFailoverTest(int delay)
         {
             _log.Debug("void DoFailoverTest(IConnectionInfo connectionInfo): called");

             // Wait for all of the test messages to be received, checking that this occurs within the test time limit.
 	    bool withinTimeout = false;

             for (int i = 1; i <= NUM_MESSAGES; ++i)
             {
 		SendMessage();

 		// Prompt the user to cause a failure if at the fail point.
 		if (i == FAIL_POINT)
 		{
 		    for( int min = delay ; min > 0 ; min--)
 		    {
 		       Console.WriteLine("Waiting for "+min+" minutes to test connection time bug.");
 		       Thread.Sleep(60*1000);
 		    }

 		    PromptAndWait("Cause a broker failure now, then press return.");
 		    Console.WriteLine("NOTE: ensure that the delay between killing the broker and continuing here is less than 20 second");

 		    Console.WriteLine("Sending a message to ensure send right after works");

 		    SendMessage();

 		    Console.WriteLine("Waiting for fail-over to complete before continuing...");


 		    lock(failoverComplete)
 		    {
 			if (!failedOver)
 			{
 			    withinTimeout = Monitor.Wait(failoverComplete, TIMEOUT);
 			}
 			else
 			{
 			    withinTimeout=true;
 			}
 		    }

 		    if (!withinTimeout)
 		    {
 			PromptAndWait("Failover has not yet occured. Press enter to give up waiting.");
 		    }
 		}
 	    }

             lock(testComplete)
             {
                 withinTimeout = Monitor.Wait(testComplete, TIMEOUT);
             }

             if (!withinTimeout)
             {
                 Assert.Fail("Test timed out, before all messages received.");
             }

             _log.Debug("void DoFailoverTest(IConnectionInfo connectionInfo): exiting");
         }

 	[Test]
         public void Test5MinuteWait()
 	{
 	    String clientId = "failover" + DateTime.Now.Ticks;

 	    QpidConnectionInfo connectionInfo = new QpidConnectionInfo();
 	    connectionInfo.Username = "guest";
 	    connectionInfo.Password = "guest";
 	    connectionInfo.ClientName = clientId;
 	    connectionInfo.VirtualHost = "/test";
 	    connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 9672, false));
 	    connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 9673, false));

 	    Init(connectionInfo);
 	    DoFailoverTest(5);
 	}

 	void SendMessage()
 	{
 	    ITextMessage msg = publishingChannel.CreateTextMessage("message=" + messagesSent);

 	    publisher.Send(msg);
 	    messagesSent++;

 	    if (transacted)
 	    {
 		publishingChannel.Commit();
 	    }

 	    Console.WriteLine("messagesSent = " + messagesSent);
 	}

         /// <summary>
         /// Receives all of the test messages.
         /// </summary>
         ///
         /// <param name="message">The newly arrived test message.</param>
         public void OnMessage(IMessage message)
         {
             try
             {
                 if (_acknowledgeMode == AcknowledgeMode.ClientAcknowledge)
                 {
                     message.Acknowledge();
                 }

                 messagesReceived++;

                 _log.Debug("messagesReceived = " + messagesReceived);

                 // Check if all of the messages in the test have been received, in which case notify the message producer that the test has
                 // succesfully completed.
                 if (messagesReceived == NUM_MESSAGES + _extraMessage)
                 {
                     lock (testComplete)
                     {
 			failedOver = true;
                         Monitor.Pulse(testComplete);
                     }
                 }
             }
             catch (QpidException e)
             {
                 _log.Fatal("Exception received. About to stop.", e);
                 Stop();
             }
         }

         /// <summary>Prompts the user on stdout and waits for a reply on stdin, using the specified prompt message.</summary>
         ///
         /// <param name="message">The message to prompt the user with.</param>
         private void PromptAndWait(string message)
         {
             Console.WriteLine("\n" + message);
             Console.ReadLine();
         }

         // <summary>Closes the test connection.</summary>
         private void Stop()
         {
             _log.Debug("Stopping...");
             try
             {
                 _connection.Close();
             }
             catch (QpidException e)
             {
                 _log.Debug("Failed to shutdown: ", e);
             }
         }

         /// <summary>
         /// Called when bytes have been transmitted to the server
         /// </summary>
         ///
         /// <param>count the number of bytes sent in total since the connection was opened</param>
         public void BytesSent(long count) {}

         /// <summary>
         /// Called when some bytes have been received on a connection
         /// </summary>
         ///
         /// <param>count the number of bytes received in total since the connection was opened</param>
         public void BytesReceived(long count) {}

         /// <summary>
         /// Called after the infrastructure has detected that failover is required but before attempting failover.
         /// </summary>
         ///
         /// <param>redirect true if the broker requested redirect. false if failover is occurring due to a connection error.</param>
         ///
         /// <return>true to continue failing over, false to veto failover and raise a connection exception</return>
         public bool PreFailover(bool redirect)
         {
             _log.Debug("public bool PreFailover(bool redirect): called");
             return true;
         }

         /// <summary>
         /// Called after connection has been made to another broker after failover has been started but before
         /// any resubscription has been done.
         /// </summary>
         ///
         /// <return> true to continue with resubscription, false to prevent automatic resubscription. This is useful in
         /// cases where the application wants to handle resubscription. Note that in the latter case all sessions, producers
         /// and consumers are invalidated.
         /// </return>
         public bool PreResubscribe()
         {
             _log.Debug("public bool PreResubscribe(): called");
             return true;
         }

         /// <summary>
         /// Called once failover has completed successfully. This is called irrespective of whether the client has
         /// vetoed automatic resubscription.
         /// </summary>
         public void FailoverComplete()
         {
 	    failedOver = true;
             _log.Debug("public void FailoverComplete(): called");
 	    Console.WriteLine("public void FailoverComplete(): called");
 	    lock (failoverComplete)
 	    {
 	      Monitor.Pulse(failoverComplete);
 	    }
         }
     }
 }
	/*
	*
	* 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 System.Runtime.InteropServices;
	using System.Threading;
	using log4net;
	using NUnit.Framework;
	using Apache.Qpid.Client.Qms;
	using Apache.Qpid.Client;
	using Apache.Qpid.Messaging;

	namespace Apache.Qpid.Integration.Tests.interactive
	{
	[TestFixture, Category("Interactive")]
	public class FailoverTest : IConnectionListener
	{
	private static readonly ILog _log = LogManager.GetLogger(typeof(FailoverTest));

	/// <summary>Specifies the number of times to run the test cycle.</summary>
	const int NUM_MESSAGES = 10;

	/// <summary>Determines how many messages to send within each commit.</summary>
	const int COMMIT_BATCH_SIZE = 1;

	/// <summary>Specifies the duration of the pause to place between each message sent in the test.</summary>
	//const int SLEEP_MILLIS = 1;

	/// <summary>Specified the maximum time in milliseconds to wait for the test to complete.</summary>
	const int TIMEOUT = 10000;

	/// <summary>Defines the number of test messages to send, before prompting the user to fail a broker.</summary>
	const int FAIL_POINT = 5;

	/// <summary>Specified the ack mode to use for the test.</summary>
	AcknowledgeMode _acknowledgeMode = AcknowledgeMode.AutoAcknowledge;

	/// <summary>Determines whether this test runs transactionally or not. </summary>
	bool transacted = false;

	/// <summary>Holds the connection to run the test over.</summary>
	AMQConnection _connection;

	/// <summary>Holds the channel for the test message publisher. </summary>
	IChannel publishingChannel;

	/// <summary>Holds the test message publisher. </summary>
	IMessagePublisher publisher;

	/// <summary>Used to keep count of the number of messages sent. </summary>
	int messagesSent;

	/// <summary>Used to keep count of the number of messages received. </summary>
	int messagesReceived;

	/// <summary>Used to wait for test completion on. </summary>
	private static object testComplete = new Object();

	/// <summary>Used to wait for failover completion on. </summary>
	private static object failoverComplete = new Object();

	bool failedOver=false;

	/// <summary>Used to record the extra message count (1) if the message sent right after failover actually made it to the new broker.</summary>
	int _extraMessage = 0;

	/// <summary>
	/// Creates the test connection with a fail-over set up, and a producer/consumer pair on that connection.
	/// </summary>
	/// [SetUp]
	public void Init(IConnectionInfo connectionInfo)
	{
	//log4net.Config.BasicConfigurator.Configure();
	// Reset all counts.
	messagesSent = 0;
	messagesReceived = 0;
	failedOver=false;
	_extraMessage = 0;

	PromptAndWait("Ensure both brokers are running, then press Enter");

	// Create a connection for the test.
	_connection = new AMQConnection(connectionInfo);
	_connection.ConnectionListener = this;

	// Create a consumer to receive the test messages.
	IChannel receivingChannel = _connection.CreateChannel(false, _acknowledgeMode);

	string queueName = receivingChannel.GenerateUniqueName();
	receivingChannel.DeclareQueue(queueName, false, true, true);
	receivingChannel.Bind(queueName, "amq.direct", queueName);

	IMessageConsumer consumer = receivingChannel.CreateConsumerBuilder(queueName)
	.WithPrefetchLow(30)
	.WithPrefetchHigh(60).Create();

	consumer.OnMessage = new MessageReceivedDelegate(OnMessage);
	_connection.Start();

	// Create a publisher to send the test messages.
	publishingChannel = _connection.CreateChannel(transacted, AcknowledgeMode.NoAcknowledge);
	publisher = publishingChannel.CreatePublisherBuilder()
	.WithRoutingKey(queueName)
	.Create();

	_log.Debug("connection = " + _connection);
	_log.Debug("connectionInfo = " + connectionInfo);
	_log.Debug("connection.AsUrl = " + _connection.toURL());
	_log.Debug("AcknowledgeMode is " + _acknowledgeMode);
	}

	/// <summary>
	/// Clean up the test connection.
	/// </summary>
	[TearDown]
	public virtual void Shutdown()
	{
	if (!failedOver)
	{
	Assert.Fail("The failover callback never occured.");
	}

	Console.WriteLine("Test done shutting down");
	Thread.Sleep(2000);
	_connection.Close();
	}

	/// <summary>
	/// Runs a failover test, building up the connection information from its component parts. In particular the brokers
	/// to fail between are seperately added into the connection info.
	/// </summary>
	/*[Test]
	public void TestWithBasicInfo()
	{
	_log.Debug("public void TestWithBasicInfo(): called");

	// Manually create the connection parameters.
	QpidConnectionInfo connectionInfo = new QpidConnectionInfo();
	connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 5672, false));
	connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 5673, false));

	Init(connectionInfo);
	DoFailoverTest();
	}*/

	/// <summary>
	/// Runs a failover test, with the failover configuration specified in the Qpid connection URL format.
	/// </summary>
	[Test]
	public void TestWithUrl()
	{
	_log.Debug("public void runTestWithUrl(): called");

	// Parse the connection parameters from a URL.
	String clientId = "failover" + DateTime.Now.Ticks;
	string defaultUrl = "amqp://guest:guest@" + clientId + "/test" +
	"?brokerlist='tcp://localhost:9672;tcp://localhost:9673'&failover='roundrobin'";
	IConnectionInfo connectionInfo = QpidConnectionInfo.FromUrl(defaultUrl);

	Init(connectionInfo);
	DoFailoverTest(0);
	}

	/// <summary>
	/// Send the test messages, prompting at the fail point for the user to cause a broker failure. The test checks that all messages sent
	/// are received within the test time limit.
	/// </summary>
	///
	/// <param name="connectionInfo">The connection parameters, specifying the brokers to fail between.</param>
	void DoFailoverTest(int delay)
	{
	_log.Debug("void DoFailoverTest(IConnectionInfo connectionInfo): called");

	// Wait for all of the test messages to be received, checking that this occurs within the test time limit.
	bool withinTimeout = false;

	for (int i = 1; i <= NUM_MESSAGES; ++i)
	{
	SendMessage();

	// Prompt the user to cause a failure if at the fail point.
	if (i == FAIL_POINT)
	{
	for( int min = delay ; min > 0 ; min--)
	{
	Console.WriteLine("Waiting for "+min+" minutes to test connection time bug.");
	Thread.Sleep(60*1000);
	}

	PromptAndWait("Cause a broker failure now, then press return.");
	Console.WriteLine("NOTE: ensure that the delay between killing the broker and continuing here is less than 20 second");

	Console.WriteLine("Sending a message to ensure send right after works");

	SendMessage();

	Console.WriteLine("Waiting for fail-over to complete before continuing...");


	lock(failoverComplete)
	{
	if (!failedOver)
	{
	withinTimeout = Monitor.Wait(failoverComplete, TIMEOUT);
	}
	else
	{
	withinTimeout=true;
	}
	}

	if (!withinTimeout)
	{
	PromptAndWait("Failover has not yet occured. Press enter to give up waiting.");
	}
	}
	}

	lock(testComplete)
	{
	withinTimeout = Monitor.Wait(testComplete, TIMEOUT);
	}

	if (!withinTimeout)
	{
	Assert.Fail("Test timed out, before all messages received.");
	}

	_log.Debug("void DoFailoverTest(IConnectionInfo connectionInfo): exiting");
	}

	[Test]
	public void Test5MinuteWait()
	{
	String clientId = "failover" + DateTime.Now.Ticks;

	QpidConnectionInfo connectionInfo = new QpidConnectionInfo();
	connectionInfo.Username = "guest";
	connectionInfo.Password = "guest";
	connectionInfo.ClientName = clientId;
	connectionInfo.VirtualHost = "/test";
	connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 9672, false));
	connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 9673, false));

	Init(connectionInfo);
	DoFailoverTest(5);
	}

	void SendMessage()
	{
	ITextMessage msg = publishingChannel.CreateTextMessage("message=" + messagesSent);

	publisher.Send(msg);
	messagesSent++;

	if (transacted)
	{
	publishingChannel.Commit();
	}

	Console.WriteLine("messagesSent = " + messagesSent);
	}

	/// <summary>
	/// Receives all of the test messages.
	/// </summary>
	///
	/// <param name="message">The newly arrived test message.</param>
	public void OnMessage(IMessage message)
	{
	try
	{
	if (_acknowledgeMode == AcknowledgeMode.ClientAcknowledge)
	{
	message.Acknowledge();
	}

	messagesReceived++;

	_log.Debug("messagesReceived = " + messagesReceived);

	// Check if all of the messages in the test have been received, in which case notify the message producer that the test has
	// succesfully completed.
	if (messagesReceived == NUM_MESSAGES + _extraMessage)
	{
	lock (testComplete)
	{
	failedOver = true;
	Monitor.Pulse(testComplete);
	}
	}
	}
	catch (QpidException e)
	{
	_log.Fatal("Exception received. About to stop.", e);
	Stop();
	}
	}

	/// <summary>Prompts the user on stdout and waits for a reply on stdin, using the specified prompt message.</summary>
	///
	/// <param name="message">The message to prompt the user with.</param>
	private void PromptAndWait(string message)
	{
	Console.WriteLine("\n" + message);
	Console.ReadLine();
	}

	// <summary>Closes the test connection.</summary>
	private void Stop()
	{
	_log.Debug("Stopping...");
	try
	{
	_connection.Close();
	}
	catch (QpidException e)
	{
	_log.Debug("Failed to shutdown: ", e);
	}
	}

	/// <summary>
	/// Called when bytes have been transmitted to the server
	/// </summary>
	///
	/// <param>count the number of bytes sent in total since the connection was opened</param>
	public void BytesSent(long count) {}

	/// <summary>
	/// Called when some bytes have been received on a connection
	/// </summary>
	///
	/// <param>count the number of bytes received in total since the connection was opened</param>
	public void BytesReceived(long count) {}

	/// <summary>
	/// Called after the infrastructure has detected that failover is required but before attempting failover.
	/// </summary>
	///
	/// <param>redirect true if the broker requested redirect. false if failover is occurring due to a connection error.</param>
	///
	/// <return>true to continue failing over, false to veto failover and raise a connection exception</return>
	public bool PreFailover(bool redirect)
	{
	_log.Debug("public bool PreFailover(bool redirect): called");
	return true;
	}

	/// <summary>
	/// Called after connection has been made to another broker after failover has been started but before
	/// any resubscription has been done.
	/// </summary>
	///
	/// <return> true to continue with resubscription, false to prevent automatic resubscription. This is useful in
	/// cases where the application wants to handle resubscription. Note that in the latter case all sessions, producers
	/// and consumers are invalidated.
	/// </return>
	public bool PreResubscribe()
	{
	_log.Debug("public bool PreResubscribe(): called");
	return true;
	}

	/// <summary>
	/// Called once failover has completed successfully. This is called irrespective of whether the client has
	/// vetoed automatic resubscription.
	/// </summary>
	public void FailoverComplete()
	{
	failedOver = true;
	_log.Debug("public void FailoverComplete(): called");
	Console.WriteLine("public void FailoverComplete(): called");
	lock (failoverComplete)
	{
	Monitor.Pulse(failoverComplete);
	}
	}
	}
	}