RC9/qpid/dotnet/Qpid.Client/Client/AMQConnection.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.Collections;
 using System.IO;
 using System.Reflection;
 using System.Threading;
 using log4net;
 using Apache.Qpid.Client.Failover;
 using Apache.Qpid.Client.Protocol;
 using Apache.Qpid.Client.Qms;
 using Apache.Qpid.Client.State;
 using Apache.Qpid.Client.Transport;
 using Apache.Qpid.Client.Transport.Socket.Blocking;
 using Apache.Qpid.Collections;
 using Apache.Qpid.Framing;
 using Apache.Qpid.Messaging;

 namespace Apache.Qpid.Client
 {
     public class AMQConnection : Closeable, IConnection
     {
         private static readonly ILog _log = LogManager.GetLogger(typeof(AMQConnection));

         IConnectionInfo _connectionInfo;
         private int _nextChannelId = 0;

         // _Connected should be refactored with a suitable wait object.
         private bool _connected;

         Thread _heartBeatThread;
         HeartBeatThread _heartBeatRunner;

         // The last error code that occured on the connection. Used to return the correct exception to the client
         private AMQException _lastAMQException = null;

         /**
          * This is the "root" mutex that must be held when doing anything that could be impacted by failover.
          * This must be held by any child objects of this connection such as the session, producers and consumers.
          */
         private readonly Object _failoverMutex = new Object();
         public object FailoverMutex
         {
             get { return _failoverMutex; }
         }

         /**
          * Policy dictating how to failover
          */
         private FailoverPolicy _failoverPolicy;

         internal bool IsFailoverAllowed
         {
             get { if(!_connected) return false; else return _failoverPolicy.FailoverAllowed(); }
         }

         /// <summary>
         /// A channel is roughly analogous to a session. The server can negotiate the maximum number of channels
         /// per session and we must prevent the client from opening too many. Zero means unlimited.
         /// </summary>
         private ushort _maximumChannelCount;

         /// <summary>
         /// The maximum size of frame supported by the server
         /// </summary>
         private uint _maximumFrameSize;

         private AMQStateManager _stateManager;

         private AMQProtocolSession _protocolSession;
         public AMQProtocolSession ProtocolSession { get { return _protocolSession;  } }

         /// <summary>
         /// Maps from session id (Integer) to AmqChannel instance
         /// </summary>
         private readonly IDictionary _sessions = new LinkedHashtable();

         private ExceptionListenerDelegate _exceptionListener;

         private IConnectionListener _connectionListener;

         private ITransport _transport;
         public ITransport Transport { get { return _transport; } }

         /// <summary>
         /// Whether this connection is started, i.e. whether messages are flowing to consumers. It has no meaning for
         /// message publication.
         /// </summary>
         private bool _started;

         private AMQProtocolListener _protocolListener;
         public AMQProtocolListener ProtocolListener { get { return _protocolListener;  } }

         public IProtocolWriter ProtocolWriter
         {
             get { return _transport.ProtocolWriter; }
         }

         ProtocolWriter _protocolWriter;

         public ProtocolWriter ConvenientProtocolWriter
         {
             get { return _protocolWriter; }
         }

         public AMQConnection(IConnectionInfo connectionInfo)
         {
             if (connectionInfo == null)
             {
                 throw new ArgumentException("ConnectionInfo must be specified");
             }
             _log.Debug("ConnectionInfo: " + connectionInfo);
             _connectionInfo = connectionInfo;
             _log.Debug("password = " + _connectionInfo.Password);
             _failoverPolicy = new FailoverPolicy(connectionInfo);

             // We are not currently connected.
             _connected = false;

             Exception lastException = null;
             do
             {
                 try
                 {
                     IBrokerInfo brokerInfo = _failoverPolicy.GetNextBrokerInfo();
                     _log.Debug("Connecting to " + brokerInfo);
                     MakeBrokerConnection(brokerInfo);
                     break;
                 }
                 catch (Exception e)
                 {
                     lastException = e;
                     _log.Error("Unable to connect to broker " + _failoverPolicy.GetCurrentBrokerInfo(), e);
                     // XXX: Should perhaps break out of the do/while here if not a SocketException...
                 }
             } while (!_connected && _failoverPolicy.FailoverAllowed());

             _log.Debug("Are we connected:" + _connected);

             if (!_connected)
             {
             	if ( lastException is AMQException )
             	{
             		throw lastException;
             	}
             	else
             	{
             		throw new AMQConnectionException("Unable to connect", lastException);
             	}
             }

         }

         /*private ITransport LoadTransportFromAssembly(string host, int port, String assemblyName, String transportType)
         {
             //Assembly assembly = Assembly.LoadFrom(assemblyName);
             Assembly assembly = Assembly.Load(assemblyName);

             foreach (Type type in assembly.GetTypes())
             {
                 _log.Debug(String.Format("type = {0}", type));
             }

             Type transport = assembly.GetType(transportType);

             if (transport == null)
             {
                 throw new ArgumentException(
                     String.Format("Type is not found in assembly. Type={0} Assembly={1}", transportType, assemblyName));

             }

             _log.Debug("transport = " + transport);
             _log.Debug("ctors = " + transport.GetConstructors());

             ConstructorInfo info = transport.GetConstructors()[0];
             ITransport result = (ITransport)info.Invoke(new object[] { host, port, this });

             _log.Debug("transport = " + result);

             return result;
         }*/

         public void Disconnect()
         {
             _transport.Close();
         }

         #region IConnection Members

         public string ClientID
         {
             get
             {
                 CheckNotClosed();
                 return _connectionInfo.ClientName;
             }
             set
             {
                 CheckNotClosed();
                 _connectionInfo.ClientName = value;
             }
         }

         public override void Close()
         {
             lock (FailoverMutex)
             {
                 // atomically set to closed and check the _previous value was NOT CLOSED
                 if (Interlocked.Exchange(ref _closed, CLOSED) == NOT_CLOSED)
                 {
                     try
                     {
                         CloseAllSessions(null);
                         CloseConnection();
                     }
                     catch (AMQException e)
                     {
                         throw new QpidException("Error closing connection: " + e);
                     }
                 }
             }
         }

         private void CloseConnection()
         {
             _stateManager.ChangeState(AMQState.CONNECTION_CLOSING);

             AMQFrame frame = ConnectionCloseBody.CreateAMQFrame(
                 0, 200, "Qpid.NET client is closing the connection.", 0, 0);

             ProtocolWriter.Write(frame);

             _log.Debug("Blocking for connection close ok frame");

             Disconnect();
         }

         class CreateChannelFailoverSupport : FailoverSupport
         {
             private static readonly ILog _log = LogManager.GetLogger(typeof(CreateChannelFailoverSupport));

             private bool _transacted;
             private AcknowledgeMode _acknowledgeMode;
             int _prefetchHigh;
             int _prefetchLow;
             AMQConnection _connection;

             public CreateChannelFailoverSupport(AMQConnection connection, bool transacted, AcknowledgeMode acknowledgeMode, int prefetchHigh, int prefetchLow)
             {
                 _connection = connection;
                 _transacted = transacted;
                 _acknowledgeMode = acknowledgeMode;
                 _prefetchHigh = prefetchHigh;
                 _prefetchLow = prefetchLow;
             }

             protected override object operation()
             {
                 ushort channelId = _connection.NextChannelId();

                 if (_log.IsDebugEnabled)
                 {
                     _log.Debug("Write channel open frame for channel id " + channelId);
                 }

                 // We must create the channel and register it before actually sending the frame to the server to
                 // open it, so that there is no window where we could receive data on the channel and not be set
                 // up to handle it appropriately.
                 AmqChannel channel = new AmqChannel(_connection,
                         channelId, _transacted, _acknowledgeMode, _prefetchHigh, _prefetchLow);
                 _connection.ProtocolSession.AddSessionByChannel(channelId, channel);
                 _connection.RegisterSession(channelId, channel);

                 bool success = false;
                 try
                 {
                     _connection.CreateChannelOverWire(channelId, _prefetchHigh, _prefetchLow, _transacted);
                     success = true;
                 }
                 catch (AMQException e)
                 {
                     throw new QpidException("Error creating channel: " + e, e);
                 }
                 finally
                 {
                     if (!success) {
                         _connection.ProtocolSession.RemoveSessionByChannel(channelId);
                         _connection.DeregisterSession(channelId);
                     }
                 }

                 if (_connection._started)
                 {
                     channel.Start();
                 }
                 return channel;
             }
         }

         internal ushort NextChannelId()
         {
             return (ushort) Interlocked.Increment(ref _nextChannelId);
         }

         public IChannel CreateChannel(bool transacted, AcknowledgeMode acknowledgeMode)
         {
             return CreateChannel(transacted, acknowledgeMode, AmqChannel.DEFAULT_PREFETCH_HIGH_MARK);
         }

         public IChannel CreateChannel(bool transacted, AcknowledgeMode acknowledgeMode, int prefetch)
         {
            return CreateChannel(transacted, acknowledgeMode, prefetch, prefetch);
         }

         public IChannel CreateChannel(bool transacted, AcknowledgeMode acknowledgeMode, int prefetchHigh, int prefetchLow)
         {
             CheckNotClosed();
             if (ChannelLimitReached())
             {
                 throw new ChannelLimitReachedException(_maximumChannelCount);
             }
             else
             {
                 CreateChannelFailoverSupport operation =
                     new CreateChannelFailoverSupport(this, transacted, acknowledgeMode, prefetchHigh, prefetchLow);
                 return (IChannel)operation.execute(this);
             }
         }

         public void CloseSession(AmqChannel channel)
         {
             // FIXME: Don't we need FailoverSupport here (as we have SyncWrite).
             _protocolSession.CloseSession(channel);

             AMQFrame frame = ChannelCloseBody.CreateAMQFrame(
                 channel.ChannelId, 200, "JMS client closing channel", 0, 0);

             _log.Debug("Blocking for channel close frame for channel " + channel.ChannelId);
             _protocolWriter.SyncWrite(frame, typeof(ChannelCloseOkBody));
             _log.Debug("Received channel close frame");
             // When control resumes at this point, a reply will have been received that
             // indicates the broker has closed the channel successfully
         }

         public ExceptionListenerDelegate ExceptionListener
         {
             get
             {
                 CheckNotClosed();
                 return _exceptionListener;
             }
             set
             {
                 CheckNotClosed();
                 _exceptionListener = value;
             }
         }

         /// <summary>
         /// Start the connection, i.e. start flowing messages. Note that this method must be called only from a single thread
         /// and is not thread safe (which is legal according to the JMS specification).
         /// @throws JMSException
         /// </summary>
         public void Start()
         {
             CheckNotClosed();

             if (!_started)
             {
                 foreach (DictionaryEntry lde in _sessions)
                 {
                     AmqChannel s = (AmqChannel)lde.Value;
                     s.Start();
                 }
                 _started = true;
             }
         }

         public void Stop()
         {
             CheckNotClosed();

             if (_started)
             {
                 foreach (DictionaryEntry lde in _sessions)
                 {
                     AmqChannel s = (AmqChannel) lde.Value;
                     s.Stop();
                 }
                 _started = false;
             }
         }

         public IConnectionListener ConnectionListener
         {
             get { return _connectionListener; }
             set { _connectionListener = value; }
         }

         #endregion

         #region IDisposable Members

         public void Dispose()
         {
             Close();
         }

         #endregion

         private bool ChannelLimitReached()
         {
             return _maximumChannelCount != 0 && _sessions.Count == _maximumChannelCount;
         }

         /// <summary>
         /// Close all the sessions, either due to normal connection closure or due to an error occurring.
         /// @param cause if not null, the error that is causing this shutdown
         /// </summary>
         private void CloseAllSessions(Exception cause)
         {
             _log.Debug("Closing all session in connection " + this);
             ICollection sessions = new ArrayList(_sessions.Values);
             foreach (AmqChannel channel in sessions)
             {
                 _log.Debug("Closing channel " + channel);
                 if (cause != null)
                 {
                     channel.ClosedWithException(cause);
                 }
                 else
                 {
                     try
                     {
                         channel.Close();
                     }
                     catch (QpidException e)
                     {
                         _log.Error("Error closing channel: " + e);
                     }
                 }
             }
             _log.Debug("Done closing all sessions in connection " + this);
         }

         public int MaximumChannelCount
         {
             get
             {
                 CheckNotClosed();
                 return _maximumChannelCount;
             }
         }

         internal void SetMaximumChannelCount(ushort maximumChannelCount)
         {
             CheckNotClosed();
             _maximumChannelCount = maximumChannelCount;
         }

         public uint MaximumFrameSize
         {
             get
             {
                 return _maximumFrameSize;
             }

             set
             {
                 _maximumFrameSize = value;
             }
         }

         public IDictionary Sessions
         {
             get
             {
                 return _sessions;
             }
         }

         public string Host
         {
             get
             {
                 return _failoverPolicy.GetCurrentBrokerInfo().Host;
             }
         }

         public int Port
         {
             get
             {
                 return _failoverPolicy.GetCurrentBrokerInfo().Port;
             }
         }

         public string Username
         {
             get
             {
                 return _connectionInfo.Username;
             }
         }

         public string Password
         {
             get
             {
                 return _connectionInfo.Password;
             }
         }

         public string VirtualHost
         {
             get
             {
                 return _connectionInfo.VirtualHost;
             }
         }

         /// <summary>
         /// Invoked by the AMQProtocolSession when a protocol session exception has occurred.
         /// This method sends the exception to a JMS exception listener, if configured, and
         /// propagates the exception to sessions, which in turn will propagate to consumers.
         /// This allows synchronous consumers to have exceptions thrown to them.
         /// </summary>
         /// <param name="cause">the exception</param>
         public void ExceptionReceived(Exception cause)
         {
             if (_exceptionListener != null)
             {
                 // Listener expects one of these...
                 QpidException xe;

                 if (cause is QpidException)
                 {
                     xe = (QpidException) cause;
                 }
                 else
                 {
                     xe = new QpidException("Exception thrown against " + ToString() + ": " + cause, cause);
                 }
                 // in the case of an IOException, MINA has closed the protocol session so we set _closed to true
                 // so that any generic client code that tries to close the connection will not mess up this error
                 // handling sequence
                 if (cause is IOException)
                 {
                     Interlocked.Exchange(ref _closed, CLOSED);
                 }
 #if __MonoCS__
                 _exceptionListener(xe);
 #else
                 _exceptionListener.Invoke(xe);
 #endif
             }
             else
             {
                 _log.Error("Connection exception: " + cause);
             }

             // An undelivered is not fatal to the connections usability.
             if (!(cause is AMQUndeliveredException))
             {
                 Interlocked.Exchange(ref _closed, CLOSED);
                 CloseAllSessions(cause);
             }
             else
             {
                 ;
             }
         }

         internal void RegisterSession(int channelId, AmqChannel channel)
         {
             _sessions[channelId] = channel;
         }

         internal void DeregisterSession(int channelId)
         {
             _sessions.Remove(channelId);
         }

         /**
          * Fire the preFailover event to the registered connection listener (if any)
          *
          * @param redirect true if this is the result of a redirect request rather than a connection error
          * @return true if no listener or listener does not veto change
          */
         public bool FirePreFailover(bool redirect)
         {
             bool proceed = true;
             if (_connectionListener != null)
             {
                 proceed = _connectionListener.PreFailover(redirect);
             }
             return proceed;
         }

         /**
          * Fire the preResubscribe event to the registered connection listener (if any). If the listener
          * vetoes resubscription then all the sessions are closed.
          *
          * @return true if no listener or listener does not veto resubscription.
          * @throws JMSException
          */
         public bool FirePreResubscribe()
         {
             if (_connectionListener != null)
             {
                 bool resubscribe = _connectionListener.PreResubscribe();
                 if (!resubscribe)
                 {
                     MarkAllSessionsClosed();
                 }
                 return resubscribe;
             }
             else
             {
                 return true;
             }
         }

         /**
          * Marks all sessions and their children as closed without sending any protocol messages. Useful when
          * you need to mark objects "visible" in userland as closed after failover or other significant event that
          * impacts the connection.
          * <p/>
          * The caller must hold the failover mutex before calling this method.
          */
         private void MarkAllSessionsClosed()
         {
             //LinkedList sessionCopy = new LinkedList(_sessions.values());
             ArrayList sessionCopy = new ArrayList(_sessions.Values);
             foreach (AmqChannel session in sessionCopy)
             {
                 session.MarkClosed();
             }
             _sessions.Clear();
         }

         /**
          * Fires a failover complete event to the registered connection listener (if any).
          */
         public void FireFailoverComplete()
         {
             if (_connectionListener != null)
             {
                 _connectionListener.FailoverComplete();
             }
         }

         public bool AttemptReconnection(String host, int port, SslOptions sslConfig)
         {
             IBrokerInfo bd = new AmqBrokerInfo("amqp", host, port, sslConfig);

             _failoverPolicy.setBroker(bd);

             try
             {
                 MakeBrokerConnection(bd);
                 return true;
             }
             catch (Exception e)
             {
                 _log.Debug("Unable to connect to broker at " + bd, e);
                 AttemptReconnection();
             }
             return false;
         }

         private void MakeBrokerConnection(IBrokerInfo brokerDetail)
         {
             try
             {
                 _stateManager = new AMQStateManager();
                 _protocolListener = new AMQProtocolListener(this, _stateManager);
                 _protocolListener.AddFrameListener(_stateManager);

                 /*
                 // Currently there is only one transport option - BlockingSocket.
                 String assemblyName = "Apache.Qpid.Client.Transport.Socket.Blocking.dll";
                 String transportType = "Apache.Qpid.Client.Transport.Socket.Blocking.BlockingSocketTransport";

                 // Load the transport assembly dynamically.
                 _transport = LoadTransportFromAssembly(brokerDetail.getHost(), brokerDetail.getPort(), assemblyName, transportType);
                 */

                 _transport = new BlockingSocketTransport();

                 // Connect.
                 _transport.Connect(brokerDetail, this);
                 _protocolWriter = new ProtocolWriter(_transport.ProtocolWriter, _protocolListener);
                 _protocolSession = new AMQProtocolSession(_transport.ProtocolWriter, _transport, this);
                 _protocolListener.ProtocolSession = _protocolSession;

                 // Now start the connection "handshake".
                 _transport.ProtocolWriter.Write(new ProtocolInitiation());

                 // Blocks until the connection has been opened.
                 _stateManager.AttainState(AMQState.CONNECTION_OPEN);

                 _failoverPolicy.attainedConnection();

                 // XXX: Again this should be changed to a suitable notify.
                 _connected = true;
             }
             catch (AMQException e)
             {
                 _lastAMQException = e;
                 throw; // rethrow
             }
         }

         public bool AttemptReconnection()
         {
             while (_failoverPolicy.FailoverAllowed())
             {
                 try
                 {
                     MakeBrokerConnection(_failoverPolicy.GetNextBrokerInfo());
                     return true;
                 }
                 catch (Exception e)
                 {
                     if (!(e is AMQException))
                     {
                         _log.Debug("Unable to connect to broker at " + _failoverPolicy.GetCurrentBrokerInfo(), e);
                     }
                     else
                     {
                         _log.Debug(e.Message + ":Unable to connect to broker at " + _failoverPolicy.GetCurrentBrokerInfo());
                     }
                 }
             }

             // Connection unsuccessful.
             return false;
         }

         /**
          * For all channels, and for all consumers in those sessions, resubscribe. This is called during failover handling.
          * The caller must hold the failover mutex before calling this method.
          */
         public void ResubscribeChannels()
         {
             ArrayList channels = new ArrayList(_sessions.Values);
             _log.Debug(String.Format("Resubscribing sessions = {0} sessions.size={1}", channels, channels.Count));
             foreach (AmqChannel channel in channels)
             {
                 _protocolSession.AddSessionByChannel(channel.ChannelId, channel);
                 ReopenChannel(
                    channel.ChannelId,
                    channel.DefaultPrefetchHigh,
                    channel.DefaultPrefetchLow,
                    channel.Transacted
                    );
                 channel.ReplayOnFailOver();
             }
         }

         private void ReopenChannel(ushort channelId, int prefetchHigh, int prefetchLow, bool transacted)
         {
             _log.Debug(string.Format("Reopening channel id={0} prefetchHigh={1} prefetchLow={2} transacted={3}",
                 channelId, prefetchHigh, prefetchLow, transacted));
             try
             {
                 CreateChannelOverWire(channelId, prefetchHigh, prefetchLow, transacted);
             }
             catch (AMQException e)
             {
                 _protocolSession.RemoveSessionByChannel(channelId);
                 DeregisterSession(channelId);
                 throw new AMQException("Error reopening channel " + channelId + " after failover: " + e);
             }
         }

         void CreateChannelOverWire(ushort channelId, int prefetchHigh, int prefetchLow, bool transacted)
         {
             _protocolWriter.SyncWrite(ChannelOpenBody.CreateAMQFrame(channelId, null), typeof (ChannelOpenOkBody));

             // Don't use the BasicQos frame if connecting to OpenAMQ (at it is not support). We
             // know this when we connection using AMQP 0.7
             if (ProtocolInitiation.CURRENT_PROTOCOL_VERSION_MAJOR != 7)
             {
                 // Basic.Qos frame appears to not be supported by OpenAMQ 1.0d.
                 _protocolWriter.SyncWrite(BasicQosBody.CreateAMQFrame(channelId, 0, (ushort)prefetchHigh, false), typeof (BasicQosOkBody));
             }

             if (transacted)
             {
                 if (_log.IsDebugEnabled)
                 {
                     _log.Debug("Issuing TxSelect for " + channelId);
                 }
                 _protocolWriter.SyncWrite(TxSelectBody.CreateAMQFrame(channelId), typeof(TxSelectOkBody));
             }
         }

         public String toURL()
         {
             return _connectionInfo.AsUrl();
         }

         class HeartBeatThread
         {
             int _heartbeatMillis;
             IProtocolWriter _protocolWriter;
             bool _run = true;

             public HeartBeatThread(IProtocolWriter protocolWriter, int heartbeatMillis)
             {
                 _protocolWriter = protocolWriter;
                 _heartbeatMillis = heartbeatMillis;
             }

             public void Run()
             {
                 while (_run)
                 {
                     Thread.Sleep(_heartbeatMillis);
                     if (!_run) break;
                     _log.Debug("Sending heartbeat");
                     // TODO: Can we optimise this so that heartbeats are only written when we haven't sent anything recently to the broker?
                     _protocolWriter.Write(HeartbeatBody.FRAME);
                 }
                 _log.Debug("Heatbeat thread stopped");
             }

             public void Stop()
             {
                 _run = false;
             }
         }

         public void StartHeartBeatThread(int heartbeatSeconds)
         {
             _log.Debug("Starting new heartbeat thread");
             _heartBeatRunner = new HeartBeatThread(ProtocolWriter, heartbeatSeconds * 1000);
             _heartBeatThread = new Thread(new ThreadStart(_heartBeatRunner.Run));
             _heartBeatThread.Name = "HeartBeat";
             _heartBeatThread.Start();
         }

         public void StopHeartBeatThread()
         {
             if (_heartBeatRunner != null)
             {
                 _log.Debug("Stopping old heartbeat thread");
                 _heartBeatRunner.Stop();
             }
         }
     }
 }
	/*
	*
	* 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;
	using System.IO;
	using System.Reflection;
	using System.Threading;
	using log4net;
	using Apache.Qpid.Client.Failover;
	using Apache.Qpid.Client.Protocol;
	using Apache.Qpid.Client.Qms;
	using Apache.Qpid.Client.State;
	using Apache.Qpid.Client.Transport;
	using Apache.Qpid.Client.Transport.Socket.Blocking;
	using Apache.Qpid.Collections;
	using Apache.Qpid.Framing;
	using Apache.Qpid.Messaging;

	namespace Apache.Qpid.Client
	{
	public class AMQConnection : Closeable, IConnection
	{
	private static readonly ILog _log = LogManager.GetLogger(typeof(AMQConnection));

	IConnectionInfo _connectionInfo;
	private int _nextChannelId = 0;

	// _Connected should be refactored with a suitable wait object.
	private bool _connected;

	Thread _heartBeatThread;
	HeartBeatThread _heartBeatRunner;

	// The last error code that occured on the connection. Used to return the correct exception to the client
	private AMQException _lastAMQException = null;

	/**
	* This is the "root" mutex that must be held when doing anything that could be impacted by failover.
	* This must be held by any child objects of this connection such as the session, producers and consumers.
	*/
	private readonly Object _failoverMutex = new Object();
	public object FailoverMutex
	{
	get { return _failoverMutex; }
	}

	/**
	* Policy dictating how to failover
	*/
	private FailoverPolicy _failoverPolicy;

	internal bool IsFailoverAllowed
	{
	get { if(!_connected) return false; else return _failoverPolicy.FailoverAllowed(); }
	}

	/// <summary>
	/// A channel is roughly analogous to a session. The server can negotiate the maximum number of channels
	/// per session and we must prevent the client from opening too many. Zero means unlimited.
	/// </summary>
	private ushort _maximumChannelCount;

	/// <summary>
	/// The maximum size of frame supported by the server
	/// </summary>
	private uint _maximumFrameSize;

	private AMQStateManager _stateManager;

	private AMQProtocolSession _protocolSession;
	public AMQProtocolSession ProtocolSession { get { return _protocolSession; } }

	/// <summary>
	/// Maps from session id (Integer) to AmqChannel instance
	/// </summary>
	private readonly IDictionary _sessions = new LinkedHashtable();

	private ExceptionListenerDelegate _exceptionListener;

	private IConnectionListener _connectionListener;

	private ITransport _transport;
	public ITransport Transport { get { return _transport; } }

	/// <summary>
	/// Whether this connection is started, i.e. whether messages are flowing to consumers. It has no meaning for
	/// message publication.
	/// </summary>
	private bool _started;

	private AMQProtocolListener _protocolListener;
	public AMQProtocolListener ProtocolListener { get { return _protocolListener; } }

	public IProtocolWriter ProtocolWriter
	{
	get { return _transport.ProtocolWriter; }
	}

	ProtocolWriter _protocolWriter;

	public ProtocolWriter ConvenientProtocolWriter
	{
	get { return _protocolWriter; }
	}

	public AMQConnection(IConnectionInfo connectionInfo)
	{
	if (connectionInfo == null)
	{
	throw new ArgumentException("ConnectionInfo must be specified");
	}
	_log.Debug("ConnectionInfo: " + connectionInfo);
	_connectionInfo = connectionInfo;
	_log.Debug("password = " + _connectionInfo.Password);
	_failoverPolicy = new FailoverPolicy(connectionInfo);

	// We are not currently connected.
	_connected = false;

	Exception lastException = null;
	do
	{
	try
	{
	IBrokerInfo brokerInfo = _failoverPolicy.GetNextBrokerInfo();
	_log.Debug("Connecting to " + brokerInfo);
	MakeBrokerConnection(brokerInfo);
	break;
	}
	catch (Exception e)
	{
	lastException = e;
	_log.Error("Unable to connect to broker " + _failoverPolicy.GetCurrentBrokerInfo(), e);
	// XXX: Should perhaps break out of the do/while here if not a SocketException...
	}
	} while (!_connected && _failoverPolicy.FailoverAllowed());

	_log.Debug("Are we connected:" + _connected);

	if (!_connected)
	{
	if ( lastException is AMQException )
	{
	throw lastException;
	}
	else
	{
	throw new AMQConnectionException("Unable to connect", lastException);
	}
	}

	}

	/*private ITransport LoadTransportFromAssembly(string host, int port, String assemblyName, String transportType)
	{
	//Assembly assembly = Assembly.LoadFrom(assemblyName);
	Assembly assembly = Assembly.Load(assemblyName);

	foreach (Type type in assembly.GetTypes())
	{
	_log.Debug(String.Format("type = {0}", type));
	}

	Type transport = assembly.GetType(transportType);

	if (transport == null)
	{
	throw new ArgumentException(
	String.Format("Type is not found in assembly. Type={0} Assembly={1}", transportType, assemblyName));

	}

	_log.Debug("transport = " + transport);
	_log.Debug("ctors = " + transport.GetConstructors());

	ConstructorInfo info = transport.GetConstructors()[0];
	ITransport result = (ITransport)info.Invoke(new object[] { host, port, this });

	_log.Debug("transport = " + result);

	return result;
	}*/

	public void Disconnect()
	{
	_transport.Close();
	}

	#region IConnection Members

	public string ClientID
	{
	get
	{
	CheckNotClosed();
	return _connectionInfo.ClientName;
	}
	set
	{
	CheckNotClosed();
	_connectionInfo.ClientName = value;
	}
	}

	public override void Close()
	{
	lock (FailoverMutex)
	{
	// atomically set to closed and check the _previous value was NOT CLOSED
	if (Interlocked.Exchange(ref _closed, CLOSED) == NOT_CLOSED)
	{
	try
	{
	CloseAllSessions(null);
	CloseConnection();
	}
	catch (AMQException e)
	{
	throw new QpidException("Error closing connection: " + e);
	}
	}
	}
	}

	private void CloseConnection()
	{
	_stateManager.ChangeState(AMQState.CONNECTION_CLOSING);

	AMQFrame frame = ConnectionCloseBody.CreateAMQFrame(
	0, 200, "Qpid.NET client is closing the connection.", 0, 0);

	ProtocolWriter.Write(frame);

	_log.Debug("Blocking for connection close ok frame");

	Disconnect();
	}

	class CreateChannelFailoverSupport : FailoverSupport
	{
	private static readonly ILog _log = LogManager.GetLogger(typeof(CreateChannelFailoverSupport));

	private bool _transacted;
	private AcknowledgeMode _acknowledgeMode;
	int _prefetchHigh;
	int _prefetchLow;
	AMQConnection _connection;

	public CreateChannelFailoverSupport(AMQConnection connection, bool transacted, AcknowledgeMode acknowledgeMode, int prefetchHigh, int prefetchLow)
	{
	_connection = connection;
	_transacted = transacted;
	_acknowledgeMode = acknowledgeMode;
	_prefetchHigh = prefetchHigh;
	_prefetchLow = prefetchLow;
	}

	protected override object operation()
	{
	ushort channelId = _connection.NextChannelId();

	if (_log.IsDebugEnabled)
	{
	_log.Debug("Write channel open frame for channel id " + channelId);
	}

	// We must create the channel and register it before actually sending the frame to the server to
	// open it, so that there is no window where we could receive data on the channel and not be set
	// up to handle it appropriately.
	AmqChannel channel = new AmqChannel(_connection,
	channelId, _transacted, _acknowledgeMode, _prefetchHigh, _prefetchLow);
	_connection.ProtocolSession.AddSessionByChannel(channelId, channel);
	_connection.RegisterSession(channelId, channel);

	bool success = false;
	try
	{
	_connection.CreateChannelOverWire(channelId, _prefetchHigh, _prefetchLow, _transacted);
	success = true;
	}
	catch (AMQException e)
	{
	throw new QpidException("Error creating channel: " + e, e);
	}
	finally
	{
	if (!success) {
	_connection.ProtocolSession.RemoveSessionByChannel(channelId);
	_connection.DeregisterSession(channelId);
	}
	}

	if (_connection._started)
	{
	channel.Start();
	}
	return channel;
	}
	}

	internal ushort NextChannelId()
	{
	return (ushort) Interlocked.Increment(ref _nextChannelId);
	}

	public IChannel CreateChannel(bool transacted, AcknowledgeMode acknowledgeMode)
	{
	return CreateChannel(transacted, acknowledgeMode, AmqChannel.DEFAULT_PREFETCH_HIGH_MARK);
	}

	public IChannel CreateChannel(bool transacted, AcknowledgeMode acknowledgeMode, int prefetch)
	{
	return CreateChannel(transacted, acknowledgeMode, prefetch, prefetch);
	}

	public IChannel CreateChannel(bool transacted, AcknowledgeMode acknowledgeMode, int prefetchHigh, int prefetchLow)
	{
	CheckNotClosed();
	if (ChannelLimitReached())
	{
	throw new ChannelLimitReachedException(_maximumChannelCount);
	}
	else
	{
	CreateChannelFailoverSupport operation =
	new CreateChannelFailoverSupport(this, transacted, acknowledgeMode, prefetchHigh, prefetchLow);
	return (IChannel)operation.execute(this);
	}
	}

	public void CloseSession(AmqChannel channel)
	{
	// FIXME: Don't we need FailoverSupport here (as we have SyncWrite).
	_protocolSession.CloseSession(channel);

	AMQFrame frame = ChannelCloseBody.CreateAMQFrame(
	channel.ChannelId, 200, "JMS client closing channel", 0, 0);

	_log.Debug("Blocking for channel close frame for channel " + channel.ChannelId);
	_protocolWriter.SyncWrite(frame, typeof(ChannelCloseOkBody));
	_log.Debug("Received channel close frame");
	// When control resumes at this point, a reply will have been received that
	// indicates the broker has closed the channel successfully
	}

	public ExceptionListenerDelegate ExceptionListener
	{
	get
	{
	CheckNotClosed();
	return _exceptionListener;
	}
	set
	{
	CheckNotClosed();
	_exceptionListener = value;
	}
	}

	/// <summary>
	/// Start the connection, i.e. start flowing messages. Note that this method must be called only from a single thread
	/// and is not thread safe (which is legal according to the JMS specification).
	/// @throws JMSException
	/// </summary>
	public void Start()
	{
	CheckNotClosed();

	if (!_started)
	{
	foreach (DictionaryEntry lde in _sessions)
	{
	AmqChannel s = (AmqChannel)lde.Value;
	s.Start();
	}
	_started = true;
	}
	}

	public void Stop()
	{
	CheckNotClosed();

	if (_started)
	{
	foreach (DictionaryEntry lde in _sessions)
	{
	AmqChannel s = (AmqChannel) lde.Value;
	s.Stop();
	}
	_started = false;
	}
	}

	public IConnectionListener ConnectionListener
	{
	get { return _connectionListener; }
	set { _connectionListener = value; }
	}

	#endregion

	#region IDisposable Members

	public void Dispose()
	{
	Close();
	}

	#endregion

	private bool ChannelLimitReached()
	{
	return _maximumChannelCount != 0 && _sessions.Count == _maximumChannelCount;
	}

	/// <summary>
	/// Close all the sessions, either due to normal connection closure or due to an error occurring.
	/// @param cause if not null, the error that is causing this shutdown
	/// </summary>
	private void CloseAllSessions(Exception cause)
	{
	_log.Debug("Closing all session in connection " + this);
	ICollection sessions = new ArrayList(_sessions.Values);
	foreach (AmqChannel channel in sessions)
	{
	_log.Debug("Closing channel " + channel);
	if (cause != null)
	{
	channel.ClosedWithException(cause);
	}
	else
	{
	try
	{
	channel.Close();
	}
	catch (QpidException e)
	{
	_log.Error("Error closing channel: " + e);
	}
	}
	}
	_log.Debug("Done closing all sessions in connection " + this);
	}

	public int MaximumChannelCount
	{
	get
	{
	CheckNotClosed();
	return _maximumChannelCount;
	}
	}

	internal void SetMaximumChannelCount(ushort maximumChannelCount)
	{
	CheckNotClosed();
	_maximumChannelCount = maximumChannelCount;
	}

	public uint MaximumFrameSize
	{
	get
	{
	return _maximumFrameSize;
	}

	set
	{
	_maximumFrameSize = value;
	}
	}

	public IDictionary Sessions
	{
	get
	{
	return _sessions;
	}
	}

	public string Host
	{
	get
	{
	return _failoverPolicy.GetCurrentBrokerInfo().Host;
	}
	}

	public int Port
	{
	get
	{
	return _failoverPolicy.GetCurrentBrokerInfo().Port;
	}
	}

	public string Username
	{
	get
	{
	return _connectionInfo.Username;
	}
	}

	public string Password
	{
	get
	{
	return _connectionInfo.Password;
	}
	}

	public string VirtualHost
	{
	get
	{
	return _connectionInfo.VirtualHost;
	}
	}

	/// <summary>
	/// Invoked by the AMQProtocolSession when a protocol session exception has occurred.
	/// This method sends the exception to a JMS exception listener, if configured, and
	/// propagates the exception to sessions, which in turn will propagate to consumers.
	/// This allows synchronous consumers to have exceptions thrown to them.
	/// </summary>
	/// <param name="cause">the exception</param>
	public void ExceptionReceived(Exception cause)
	{
	if (_exceptionListener != null)
	{
	// Listener expects one of these...
	QpidException xe;

	if (cause is QpidException)
	{
	xe = (QpidException) cause;
	}
	else
	{
	xe = new QpidException("Exception thrown against " + ToString() + ": " + cause, cause);
	}
	// in the case of an IOException, MINA has closed the protocol session so we set _closed to true
	// so that any generic client code that tries to close the connection will not mess up this error
	// handling sequence
	if (cause is IOException)
	{
	Interlocked.Exchange(ref _closed, CLOSED);
	}
	#if __MonoCS__
	_exceptionListener(xe);
	#else
	_exceptionListener.Invoke(xe);
	#endif
	}
	else
	{
	_log.Error("Connection exception: " + cause);
	}

	// An undelivered is not fatal to the connections usability.
	if (!(cause is AMQUndeliveredException))
	{
	Interlocked.Exchange(ref _closed, CLOSED);
	CloseAllSessions(cause);
	}
	else
	{
	;
	}
	}

	internal void RegisterSession(int channelId, AmqChannel channel)
	{
	_sessions[channelId] = channel;
	}

	internal void DeregisterSession(int channelId)
	{
	_sessions.Remove(channelId);
	}

	/**
	* Fire the preFailover event to the registered connection listener (if any)
	*
	* @param redirect true if this is the result of a redirect request rather than a connection error
	* @return true if no listener or listener does not veto change
	*/
	public bool FirePreFailover(bool redirect)
	{
	bool proceed = true;
	if (_connectionListener != null)
	{
	proceed = _connectionListener.PreFailover(redirect);
	}
	return proceed;
	}

	/**
	* Fire the preResubscribe event to the registered connection listener (if any). If the listener
	* vetoes resubscription then all the sessions are closed.
	*
	* @return true if no listener or listener does not veto resubscription.
	* @throws JMSException
	*/
	public bool FirePreResubscribe()
	{
	if (_connectionListener != null)
	{
	bool resubscribe = _connectionListener.PreResubscribe();
	if (!resubscribe)
	{
	MarkAllSessionsClosed();
	}
	return resubscribe;
	}
	else
	{
	return true;
	}
	}

	/**
	* Marks all sessions and their children as closed without sending any protocol messages. Useful when
	* you need to mark objects "visible" in userland as closed after failover or other significant event that
	* impacts the connection.
	* <p/>
	* The caller must hold the failover mutex before calling this method.
	*/
	private void MarkAllSessionsClosed()
	{
	//LinkedList sessionCopy = new LinkedList(_sessions.values());
	ArrayList sessionCopy = new ArrayList(_sessions.Values);
	foreach (AmqChannel session in sessionCopy)
	{
	session.MarkClosed();
	}
	_sessions.Clear();
	}

	/**
	* Fires a failover complete event to the registered connection listener (if any).
	*/
	public void FireFailoverComplete()
	{
	if (_connectionListener != null)
	{
	_connectionListener.FailoverComplete();
	}
	}

	public bool AttemptReconnection(String host, int port, SslOptions sslConfig)
	{
	IBrokerInfo bd = new AmqBrokerInfo("amqp", host, port, sslConfig);

	_failoverPolicy.setBroker(bd);

	try
	{
	MakeBrokerConnection(bd);
	return true;
	}
	catch (Exception e)
	{
	_log.Debug("Unable to connect to broker at " + bd, e);
	AttemptReconnection();
	}
	return false;
	}

	private void MakeBrokerConnection(IBrokerInfo brokerDetail)
	{
	try
	{
	_stateManager = new AMQStateManager();
	_protocolListener = new AMQProtocolListener(this, _stateManager);
	_protocolListener.AddFrameListener(_stateManager);

	/*
	// Currently there is only one transport option - BlockingSocket.
	String assemblyName = "Apache.Qpid.Client.Transport.Socket.Blocking.dll";
	String transportType = "Apache.Qpid.Client.Transport.Socket.Blocking.BlockingSocketTransport";

	// Load the transport assembly dynamically.
	_transport = LoadTransportFromAssembly(brokerDetail.getHost(), brokerDetail.getPort(), assemblyName, transportType);
	*/

	_transport = new BlockingSocketTransport();

	// Connect.
	_transport.Connect(brokerDetail, this);
	_protocolWriter = new ProtocolWriter(_transport.ProtocolWriter, _protocolListener);
	_protocolSession = new AMQProtocolSession(_transport.ProtocolWriter, _transport, this);
	_protocolListener.ProtocolSession = _protocolSession;

	// Now start the connection "handshake".
	_transport.ProtocolWriter.Write(new ProtocolInitiation());

	// Blocks until the connection has been opened.
	_stateManager.AttainState(AMQState.CONNECTION_OPEN);

	_failoverPolicy.attainedConnection();

	// XXX: Again this should be changed to a suitable notify.
	_connected = true;
	}
	catch (AMQException e)
	{
	_lastAMQException = e;
	throw; // rethrow
	}
	}

	public bool AttemptReconnection()
	{
	while (_failoverPolicy.FailoverAllowed())
	{
	try
	{
	MakeBrokerConnection(_failoverPolicy.GetNextBrokerInfo());
	return true;
	}
	catch (Exception e)
	{
	if (!(e is AMQException))
	{
	_log.Debug("Unable to connect to broker at " + _failoverPolicy.GetCurrentBrokerInfo(), e);
	}
	else
	{
	_log.Debug(e.Message + ":Unable to connect to broker at " + _failoverPolicy.GetCurrentBrokerInfo());
	}
	}
	}

	// Connection unsuccessful.
	return false;
	}

	/**
	* For all channels, and for all consumers in those sessions, resubscribe. This is called during failover handling.
	* The caller must hold the failover mutex before calling this method.
	*/
	public void ResubscribeChannels()
	{
	ArrayList channels = new ArrayList(_sessions.Values);
	_log.Debug(String.Format("Resubscribing sessions = {0} sessions.size={1}", channels, channels.Count));
	foreach (AmqChannel channel in channels)
	{
	_protocolSession.AddSessionByChannel(channel.ChannelId, channel);
	ReopenChannel(
	channel.ChannelId,
	channel.DefaultPrefetchHigh,
	channel.DefaultPrefetchLow,
	channel.Transacted
	);
	channel.ReplayOnFailOver();
	}
	}

	private void ReopenChannel(ushort channelId, int prefetchHigh, int prefetchLow, bool transacted)
	{
	_log.Debug(string.Format("Reopening channel id={0} prefetchHigh={1} prefetchLow={2} transacted={3}",
	channelId, prefetchHigh, prefetchLow, transacted));
	try
	{
	CreateChannelOverWire(channelId, prefetchHigh, prefetchLow, transacted);
	}
	catch (AMQException e)
	{
	_protocolSession.RemoveSessionByChannel(channelId);
	DeregisterSession(channelId);
	throw new AMQException("Error reopening channel " + channelId + " after failover: " + e);
	}
	}

	void CreateChannelOverWire(ushort channelId, int prefetchHigh, int prefetchLow, bool transacted)
	{
	_protocolWriter.SyncWrite(ChannelOpenBody.CreateAMQFrame(channelId, null), typeof (ChannelOpenOkBody));

	// Don't use the BasicQos frame if connecting to OpenAMQ (at it is not support). We
	// know this when we connection using AMQP 0.7
	if (ProtocolInitiation.CURRENT_PROTOCOL_VERSION_MAJOR != 7)
	{
	// Basic.Qos frame appears to not be supported by OpenAMQ 1.0d.
	_protocolWriter.SyncWrite(BasicQosBody.CreateAMQFrame(channelId, 0, (ushort)prefetchHigh, false), typeof (BasicQosOkBody));
	}

	if (transacted)
	{
	if (_log.IsDebugEnabled)
	{
	_log.Debug("Issuing TxSelect for " + channelId);
	}
	_protocolWriter.SyncWrite(TxSelectBody.CreateAMQFrame(channelId), typeof(TxSelectOkBody));
	}
	}

	public String toURL()
	{
	return _connectionInfo.AsUrl();
	}

	class HeartBeatThread
	{
	int _heartbeatMillis;
	IProtocolWriter _protocolWriter;
	bool _run = true;

	public HeartBeatThread(IProtocolWriter protocolWriter, int heartbeatMillis)
	{
	_protocolWriter = protocolWriter;
	_heartbeatMillis = heartbeatMillis;
	}

	public void Run()
	{
	while (_run)
	{
	Thread.Sleep(_heartbeatMillis);
	if (!_run) break;
	_log.Debug("Sending heartbeat");
	// TODO: Can we optimise this so that heartbeats are only written when we haven't sent anything recently to the broker?
	_protocolWriter.Write(HeartbeatBody.FRAME);
	}
	_log.Debug("Heatbeat thread stopped");
	}

	public void Stop()
	{
	_run = false;
	}
	}

	public void StartHeartBeatThread(int heartbeatSeconds)
	{
	_log.Debug("Starting new heartbeat thread");
	_heartBeatRunner = new HeartBeatThread(ProtocolWriter, heartbeatSeconds * 1000);
	_heartBeatThread = new Thread(new ThreadStart(_heartBeatRunner.Run));
	_heartBeatThread.Name = "HeartBeat";
	_heartBeatThread.Start();
	}

	public void StopHeartBeatThread()
	{
	if (_heartBeatRunner != null)
	{
	_log.Debug("Stopping old heartbeat thread");
	_heartBeatRunner.Stop();
	}
	}
	}
	}