/*
 * 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.Collections.Generic;
using System.Threading;
using Apache.NMS;
using Apache.NMS.Util;
using Apache.NMS.Test;
using Apache.NMS.ActiveMQ;
using Apache.NMS.ActiveMQ.Commands;
using Apache.NMS.ActiveMQ.Transport;
using Apache.NMS.ActiveMQ.Transport.Failover;
using Apache.NMS.ActiveMQ.Transport.Tcp;
using Apache.NMS.ActiveMQ.Transport.Mock;
using NUnit.Framework;

namespace Apache.NMS.ActiveMQ.Test
{
	[TestFixture]
	public class FailoverTransportTest
	{
		private List<Command> sent;
		private List<Command> received;
		private List<Exception> exceptions;

		private const int MAX_ATTEMPTS = 30;
        private const int MESSAGE_COUNT = 5;
        private Connection connection;
        private int msgCount = 5;
        private bool interrupted = false;
        private bool resumed = false;
        protected AutoResetEvent semaphore = new AutoResetEvent(false);

        int sessionIdx = 1;
		int consumerIdx = 1;
		int producerIdx = 1;

		private void OnException(ITransport transport, Exception exception)
		{
			Tracer.Debug("Test: Received Exception from Transport: " + exception);
			exceptions.Add(exception);
		}

		private void OnCommand(ITransport transport, Command command)
		{
			Tracer.DebugFormat("Test: Received Command from Transport: {0}", command);
			received.Add(command);
		}

		private void OnOutgoingCommand(ITransport transport, Command command)
		{
			Tracer.DebugFormat("FailoverTransportTest::OnOutgoingCommand - {0}", command);
			sent.Add(command);
		}

		private void OnResumed(ITransport sender)
		{
			Tracer.DebugFormat("FailoverTransportTest::OnResumed - {0}", sender.RemoteAddress);
			// Ensure the current mock transport has the correct outgoing command handler
			MockTransport mock = sender as MockTransport;
			Assert.IsNotNull(mock);
			mock.OutgoingCommand = OnOutgoingCommand;
		}

		private void OnInterrupted(ITransport sender)
		{
		}

		private void VerifyCommandHandlerSetting(ITransport transport, MockTransport mock)
		{
			// Walk the stack of wrapper transports.
			ITransport failoverTransportTarget = mock.Command.Target as ITransport;
			Assert.IsNotNull(failoverTransportTarget);
			ITransport mutexTransportTarget = failoverTransportTarget.Command.Target as ITransport;
			Assert.IsNotNull(mutexTransportTarget);
			ITransport responseCorrelatorTransportTarget = mutexTransportTarget.Command.Target as ITransport;
			Assert.IsNotNull(responseCorrelatorTransportTarget);
			Assert.AreEqual(transport.Command.Target, responseCorrelatorTransportTarget.Command.Target);
		}

		[SetUp]
		public void init()
		{
			sent = new List<Command>();
			received = new List<Command>();
			exceptions = new List<Exception>();
			sessionIdx = 1;
			consumerIdx = 1;
			producerIdx = 1;
            this.connection = null;
            this.msgCount = MESSAGE_COUNT;
            this.interrupted = false;
            this.resumed = false;
		}

		[Test]
		public void FailoverTransportCreateTest()
		{
			Uri uri = new Uri("failover:(mock://localhost:61616)?transport.randomize=false");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);

				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover, "Failed to create Transport");
				Assert.IsFalse(failover.Randomize, "Failed to properly set Randomize flag");

				transport.Start();

				Thread.Sleep(2000);
				Assert.IsTrue(failover.IsConnected, "Transport should be connected");
			}
		}

		[Test]
		public void FailoverTransportWithBackupsTest()
		{
			Uri uri = new Uri("failover:(mock://localhost:61616,mock://localhost:61618)?transport.randomize=false&transport.backup=true");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.Randomize);
				Assert.IsTrue(failover.Backup);

				transport.Start();
				Thread.Sleep(1000);
				Assert.IsTrue(failover.IsConnected);
			}
		}

		[Test]
		public void FailoverTransportCreateFailOnCreateTest()
		{
			Uri uri = new Uri("failover:(mock://localhost:61616?transport.failOnCreate=true)?" +
							  "transport.useExponentialBackOff=false&transport.maxReconnectAttempts=3&transport.initialReconnectDelay=100");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.UseExponentialBackOff);
				Assert.AreEqual(3, failover.MaxReconnectAttempts);
				Assert.AreEqual(100, failover.InitialReconnectDelay);

				transport.Start();
				Thread.Sleep(2000);
				Assert.IsNotEmpty(this.exceptions);
				Assert.IsFalse(failover.IsConnected);
			}
		}

        [Test]
        public void FailoverTransportCreateFailOnCreateTest2()
        {
            Uri uri = new Uri("failover:(mock://localhost:61616?transport.failOnCreate=true)?" +
                              "transport.useExponentialBackOff=false&transport.startupMaxReconnectAttempts=3&transport.initialReconnectDelay=100");
            FailoverTransportFactory factory = new FailoverTransportFactory();

            using(ITransport transport = factory.CreateTransport(uri))
            {
                Assert.IsNotNull(transport);
                transport.Command = OnCommand;
                transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

                FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
                Assert.IsNotNull(failover);
                Assert.IsFalse(failover.UseExponentialBackOff);
                Assert.AreEqual(3, failover.StartupMaxReconnectAttempts);
                Assert.AreEqual(100, failover.InitialReconnectDelay);

                transport.Start();
                Thread.Sleep(2000);
                Assert.IsNotEmpty(this.exceptions);
                Assert.IsFalse(failover.IsConnected);
            }
        }

		[Test]
		public void FailoverTransportFailOnSendMessageTest()
		{
			Uri uri = new Uri("failover:(mock://localhost:61616?transport.failOnCreate=true)?" +
							  "transport.useExponentialBackOff=false&transport.maxReconnectAttempts=3&transport.initialReconnectDelay=100");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.UseExponentialBackOff);
				Assert.AreEqual(3, failover.MaxReconnectAttempts);
				Assert.AreEqual(100, failover.InitialReconnectDelay);

				transport.Start();

				ActiveMQMessage message = new ActiveMQMessage();
				Assert.Throws<IOException>(delegate() { transport.Oneway(message); }, "Oneway call should block and then throw.");

				Assert.IsNotEmpty(this.exceptions);
				Assert.IsFalse(failover.IsConnected);
			}
		}

		[Test]
		public void FailoverTransportFailingBackupsTest()
		{
			Uri uri = new Uri(
				"failover:(mock://localhost:61616," +
						  "mock://localhost:61618?transport.failOnCreate=true)?transport.randomize=false&transport.backup=true");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.Randomize);
				Assert.IsTrue(failover.Backup);

				transport.Start();
				Thread.Sleep(2000);
				Assert.IsTrue(failover.IsConnected);
			}
		}

		[Test]
		public void FailoverTransportSendOnewayMessageTest()
		{
			int numMessages = 1000;
			Uri uri = new Uri("failover:(mock://localhost:61616)?transport.randomize=false");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.Randomize);

				failover.Resumed = OnResumed;

				transport.Start();
				Thread.Sleep(1000);
				Assert.IsTrue(failover.IsConnected);

				// Ensure the current mock transport has the correct outgoing command handler
				MockTransport mock = transport.Narrow(typeof(MockTransport)) as MockTransport;
				Assert.IsNotNull(mock);
				Assert.AreEqual(61616, mock.RemoteAddress.Port);

				VerifyCommandHandlerSetting(transport, mock);
				mock.OutgoingCommand = OnOutgoingCommand;

				ActiveMQMessage message = new ActiveMQMessage();
				for(int i = 0; i < numMessages; ++i)
				{
					transport.Oneway(message);
				}

				Thread.Sleep(1000);
				Assert.AreEqual(numMessages, this.sent.Count);
			}
		}

		[Test]
		public void FailoverTransportSendRequestTest()
		{
			Uri uri = new Uri("failover:(mock://localhost:61616)?transport.randomize=false");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.Randomize);

				failover.Resumed = OnResumed;

				transport.Start();
				Thread.Sleep(1000);
				Assert.IsTrue(failover.IsConnected);

				// Ensure the current mock transport has the correct outgoing command handler
				MockTransport mock = transport.Narrow(typeof(MockTransport)) as MockTransport;
				Assert.IsNotNull(mock);
				Assert.AreEqual(61616, mock.RemoteAddress.Port);
				VerifyCommandHandlerSetting(transport, mock);
				mock.OutgoingCommand = OnOutgoingCommand;

				ActiveMQMessage message = new ActiveMQMessage();
				int numMessages = 4;

				for(int i = 0; i < numMessages; ++i)
				{
					transport.Request(message);
				}

				Thread.Sleep(1000);
				Assert.AreEqual(numMessages, this.sent.Count);
			}
		}

		[Test]
		public void FailoverTransportSendOnewayFailTest()
		{
			Uri uri = new Uri(
				"failover:(mock://localhost:61616?transport.failOnSendMessage=true," +
						  "mock://localhost:61618)?transport.randomize=false");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.Randomize);

				failover.Resumed = OnResumed;

				transport.Start();
				Thread.Sleep(1000);
				Assert.IsTrue(failover.IsConnected);

				// Ensure the current mock transport has the correct outgoing command handler
				MockTransport mock = transport.Narrow(typeof(MockTransport)) as MockTransport;
				Assert.IsNotNull(mock);
				Assert.AreEqual(61616, mock.RemoteAddress.Port);
				VerifyCommandHandlerSetting(transport, mock);
				mock.OutgoingCommand = OnOutgoingCommand;

				ActiveMQMessage message = new ActiveMQMessage();
				int numMessages = 4;

				for(int i = 0; i < numMessages; ++i)
				{
					transport.Oneway(message);
					// Make sure we switched to second failover
					mock = transport.Narrow(typeof(MockTransport)) as MockTransport;
					Assert.IsNotNull(mock);
					Assert.AreEqual(61618, mock.RemoteAddress.Port);
				}

				Thread.Sleep(1000);
				Assert.AreEqual(numMessages, this.sent.Count);
			}
		}

		[Test]
		public void FailoverTransportSendOnewayTimeoutTest()
		{
			Uri uri = new Uri(
				"failover:(mock://localhost:61616?transport.failOnCreate=true)?transport.timeout=1000");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.AreEqual(1000, failover.Timeout);

				transport.Start();
				Thread.Sleep(1000);

				ActiveMQMessage message = new ActiveMQMessage();
				Assert.Throws<IOException>(delegate() { transport.Oneway(message); });
			}
		}

		[Test]
		public void FailoverTransportSendRequestFailTest()
		{
			Uri uri = new Uri(
				"failover:(mock://localhost:61616?transport.failOnSendMessage=true," +
						  "mock://localhost:61618)?transport.randomize=false");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.Randomize);

				failover.Resumed = OnResumed;

				transport.Start();
				Thread.Sleep(1000);
				Assert.IsTrue(failover.IsConnected);

				// Ensure the current mock transport has the correct outgoing command handler
				MockTransport mock = transport.Narrow(typeof(MockTransport)) as MockTransport;
				Assert.IsNotNull(mock);
				VerifyCommandHandlerSetting(transport, mock);
				mock.OutgoingCommand = OnOutgoingCommand;

				ActiveMQMessage message = new ActiveMQMessage();
				int numMessages = 4;

				for(int i = 0; i < numMessages; ++i)
				{
					transport.Request(message);
				}

				Thread.Sleep(1000);
				Assert.AreEqual(numMessages, this.sent.Count);
			}
		}

		[Test]
		public void TestFailoverTransportConnectionControlHandling()
		{
			Uri uri = new Uri("failover:(mock://localhost:61613)?transport.randomize=false");
			string connectedBrokers = "mock://localhost:61616?transport.name=Reconnected," +
									  "mock://localhost:61617?transport.name=Broker1," +
									  "mock://localhost:61618?transport.name=Broker2";
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.Randomize);

				const int MAX_ATTEMPTS = 50;

				transport.Start();
				
				for(int i = 0; i < MAX_ATTEMPTS; ++i)
				{
					if(failover.IsConnected)
					{
						break;
					}
					
					Thread.Sleep(100);
				}
				
				Assert.IsTrue(failover.IsConnected);

				// Ensure the current mock transport has the correct outgoing command handler
				MockTransport mock = transport.Narrow(typeof(MockTransport)) as MockTransport;
				Assert.IsNotNull(mock);
				Assert.AreEqual(61613, mock.RemoteAddress.Port);
				VerifyCommandHandlerSetting(transport, mock);
				mock.OutgoingCommand = OnOutgoingCommand;

				mock.InjectCommand(new ConnectionControl()
				{
					FaultTolerant = true,
					ConnectedBrokers = connectedBrokers,
					RebalanceConnection = true
				});

				// Give a bit of time for the Command to actually be processed.
				Thread.Sleep(2000);

				mock = null;
				
				for(int i = 0; i < MAX_ATTEMPTS; ++i)
				{
					mock = transport.Narrow(typeof(MockTransport)) as MockTransport;
					if(mock != null)
					{
						break;
					}
					
					Thread.Sleep(100);
				}
				
				Assert.IsNotNull(mock, "Error reconnecting to failover broker.");
				Assert.AreEqual(61616, mock.RemoteAddress.Port);
				Assert.AreEqual("Reconnected", mock.Name);
			}
		}

		[Test]
		public void TestPriorityBackupConfig() 
		{
		    Uri uri = new Uri("failover:(mock://localhost:61616,mock://localhost:61618)"+
			                  "?transport.randomize=false&transport.priorityBackup=true");

			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.Randomize, "Randomize should be false");
				Assert.IsTrue(failover.PriorityBackup, "Prioirity Backup not set.");

		    	transport.Start();

				for(int i = 0; i < MAX_ATTEMPTS; ++i)
				{
					if(failover.IsConnected)
					{
						break;
					}
					
					Thread.Sleep(100);
				}
				
				Assert.IsTrue(failover.IsConnected);
				Assert.IsTrue(failover.IsConnectedToPriority);
			}
		}

		[Test]
		public void TestPriorityBackupConfigPriorityURIsList() 
		{
		    Uri uri = new Uri("failover:(mock://localhost:61616,mock://localhost:61618)" +
			                  "?transport.randomize=false&transport.priorityBackup=true&" +
			                  "transport.priorityURIs=mock://localhost:61616,mock://localhost:61618");

			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover = transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.Randomize, "Randomize should be false");
				Assert.IsTrue(failover.PriorityBackup, "Prioirity Backup not set.");

				String priorityURIs = failover.PriorityURIs;
	            String[] tokens = priorityURIs.Split(new Char[] { ',' });
				Assert.AreEqual(2, tokens.Length, "Bad priorityURIs string: " + priorityURIs);

		    	transport.Start();

				for(int i = 0; i < MAX_ATTEMPTS; ++i)
				{
					if(failover.IsConnected)
					{
						break;
					}
					
					Thread.Sleep(100);
				}
				
				Assert.IsTrue(failover.IsConnected);
				Assert.IsTrue(failover.IsConnectedToPriority);
			}
		}

		[Test]
		public void OpenWireCommandsTest()
		{
			Uri uri = new Uri("failover:(mock://localhost:61616)?transport.randomize=false");
			FailoverTransportFactory factory = new FailoverTransportFactory();

			using(ITransport transport = factory.CreateTransport(uri))
			{
				Assert.IsNotNull(transport);
				transport.Command = OnCommand;
				transport.Exception = OnException;
				transport.Resumed = OnResumed;
				transport.Interrupted = OnInterrupted;

				FailoverTransport failover =  transport.Narrow(typeof(FailoverTransport)) as FailoverTransport;
				Assert.IsNotNull(failover);
				Assert.IsFalse(failover.Randomize);

				transport.Start();
				Thread.Sleep(1000);
				Assert.IsTrue(failover.IsConnected);

				ConnectionInfo connection = createConnection();
				transport.Request(connection);
				SessionInfo session1 = createSession(connection);
				transport.Request(session1);
				SessionInfo session2 = createSession(connection);
				transport.Request(session2);
				ConsumerInfo consumer1 = createConsumer(session1);
				transport.Request(consumer1);
				ConsumerInfo consumer2 = createConsumer(session1);
				transport.Request(consumer2);
				ConsumerInfo consumer3 = createConsumer(session2);
				transport.Request(consumer3);

				ProducerInfo producer1 = createProducer(session2);
				transport.Request(producer1);

				// Remove the Producers
				disposeOf(transport, producer1);

				// Remove the Consumers
				disposeOf(transport, consumer1);
				disposeOf(transport, consumer2);
				disposeOf(transport, consumer3);

				// Remove the Session instances.
				disposeOf(transport, session1);
				disposeOf(transport, session2);

				// Indicate that we are done.
				ShutdownInfo shutdown = new ShutdownInfo();
				transport.Oneway(shutdown);
			}
		}

		protected ConnectionInfo createConnection()
		{
			return new ConnectionInfo()
			{
				ClientId = Guid.NewGuid().ToString(),
				ConnectionId = new ConnectionId()
				{
					Value = Guid.NewGuid().ToString()
				}
			};
		}

		protected SessionInfo createSession(ConnectionInfo parent)
		{
			return new SessionInfo()
			{
				SessionId = new SessionId()
				{
					ConnectionId = parent.ConnectionId.Value,
					Value = sessionIdx++
				}
			};
		}

		protected ConsumerInfo createConsumer(SessionInfo parent)
		{
			return new ConsumerInfo()
			{
				ConsumerId = new ConsumerId()
				{
					ConnectionId = parent.SessionId.ConnectionId,
					SessionId = parent.SessionId.Value,
					Value = consumerIdx++
				}
			};
		}

		protected ProducerInfo createProducer(SessionInfo parent)
		{
			return new ProducerInfo()
			{
				ProducerId = new ProducerId()
				{
					ConnectionId = parent.SessionId.ConnectionId,
					SessionId = parent.SessionId.Value,
					Value = producerIdx++
				}
			};
		}

		protected void disposeOf(ITransport transport, SessionInfo session)
		{
			transport.Oneway(new RemoveInfo() { ObjectId = session.SessionId });
		}

		protected void disposeOf(ITransport transport, ConsumerInfo consumer)
		{
			transport.Oneway(new RemoveInfo() { ObjectId = consumer.ConsumerId });
		}

		protected void disposeOf(ITransport transport, ProducerInfo producer)
		{
			transport.Oneway(new RemoveInfo() { ObjectId = producer.ProducerId });
		}

        [Test]
        public void FailoverTransportFailOnProcessingReceivedMessageTest()
        {
            string uri = "failover:(tcp://${activemqhost}:61616)";
            IConnectionFactory factory = new ConnectionFactory(NMSTestSupport.ReplaceEnvVar(uri));
            using(connection = factory.CreateConnection() as Connection )
            {
                connection.ConnectionInterruptedListener +=
                    new ConnectionInterruptedListener(TransportInterrupted);
                connection.ConnectionResumedListener +=
                    new ConnectionResumedListener(TransportResumed);

                connection.Start();
                using(ISession session = connection.CreateSession())
                {
                    IDestination destination = session.GetQueue("Test?consumer.prefetchSize=1");
                    PurgeQueue(connection, destination);
                    PutMsgIntoQueue(session, destination);

                    using(IMessageConsumer consumer = session.CreateConsumer(destination))
                    {
                        consumer.Listener += OnMessage;
                        BreakConnection();
                        WaitForMessagesToArrive();
                    }
                }
            }

            Assert.IsTrue(this.interrupted);
            Assert.IsTrue(this.resumed);
        }

		[Test]
		public void FailStartupMaxReconnectAttempts()
		{
			// Connect to valid machine, but on invalid port that doesn't have a broker listening.
			string uri = "failover:(tcp://localhost:31313)?transport.StartupMaxReconnectAttempts=3";
			IConnectionFactory factory = new ConnectionFactory(NMSTestSupport.ReplaceEnvVar(uri));
			IConnection failConnection = factory.CreateConnection();
			try
			{
				failConnection.Start();
				Assert.Fail("Should not have connected to broker.");
			}
			catch(Apache.NMS.ActiveMQ.ConnectionClosedException)
			{
			}
			catch(Apache.NMS.NMSConnectionException)
			{
			}
			catch(Exception e)
			{
                Assert.Fail("Wrong Exception Thrown after max reconnect attempts.\n" +
				            "Exception thrown type: " + e.GetType());
			}
			finally
			{
				try
				{
					failConnection.Stop();
				}
				catch(Exception)
				{
                    Assert.Fail("Connection closed exception thrown while closing a connection.");
				}
				finally
				{
					try 
					{
						failConnection.Dispose();
					}
					catch(Exception)
					{
	                    Assert.Fail("Connection closed exception thrown while closing a connection.");
					}
				}
			}
		}
		
		public void TransportInterrupted()
        {
            this.interrupted = true;
        }

        public void TransportResumed()
        {
            this.resumed = true;
        }

        public void OnMessage(IMessage message)
        {
            var textMsg = message as ITextMessage;

            if(textMsg == null)
            {
                return;
            }

            msgCount--;

            // just process the first message for 10 seconds to give some time main thread
            // to restart ActiveMq broker
            if(msgCount == MESSAGE_COUNT - 1)
            {
                Thread.Sleep(10000);
            }

            if(msgCount == 0)
            {
                // if all messages were consumed then we are fine
                semaphore.Set();
            }
        }

        private void PutMsgIntoQueue(ISession session, IDestination destination)
        {
            using(IMessageProducer producer = session.CreateProducer(destination))
            {
                ITextMessage message = session.CreateTextMessage();
                for(int i = 0; i < msgCount; ++i)
                {
                    message.Text = "Test message " + (i + 1);
                    producer.Send(message);
                }
            }
        }

        public void PurgeQueue(IConnection conn, IDestination queue)
        {
            ISession session = conn.CreateSession();
            IMessageConsumer consumer = session.CreateConsumer(queue);
            while(consumer.Receive(TimeSpan.FromMilliseconds(500)) != null)
            {
            }
            consumer.Close();
            session.Close();
        }

        private void BreakConnection()
        {
            TcpTransport transport = this.connection.ITransport.Narrow(typeof(TcpTransport)) as TcpTransport;
            Assert.IsNotNull(transport);
            transport.Close();
        }

        protected void WaitForMessagesToArrive()
        {
            semaphore.WaitOne(30000, true);
            Assert.AreEqual(0, msgCount);
        }
	}
}