Merged revisions 537954-538078,538080-538083,538085-538097,538099-538108,538110-538239,538241-538881,538883-538906,538908-538911,538913-538921,538923-539191 via svnmerge from

https://svn.apache.org/repos/asf/incubator/qpid/branches/M2

........
  r537954 | tomasr | 2007-05-14 14:10:59 -0500 (Mon, 14 May 2007) | 4 lines
  
  * QPID-487 (Contributed by Carlos Medina) Fix QpidConnectionInfo.ToString()
  * QPID-485 (Contributed by Carlos Medina) Fix AmqBrokerInfo.Equals()
  * QPID-456 Enforce virtual host names start with '/'
........
  r538035 | tomasr | 2007-05-14 20:33:00 -0500 (Mon, 14 May 2007) | 6 lines
  
  * QPID-452 Improve message classes API
  * Add XML documentation to IChannel and IMessage
  * Add missing BrokerDetailTests
  * Add new tests for message creation and message factories
  * Fix wrong default encoding for text messages
........
  r539178 | tomasr | 2007-05-17 18:50:50 -0500 (Thu, 17 May 2007) | 6 lines
  
  * QPID-492 Fix Race condition in message decoding
  * QPID-249 Make ServiceRequestingClient and ServiceProvidingClient a single, self contained test
  * Fix incorrect exception message in Qpid.Buffers, improve tests
  * Make ContentBody use an sliced buffer to avoid extra data copy
  * Remove useless tests in Qpid.Client (Blocking IO tests)
........
  r539191 | tomasr | 2007-05-17 19:18:26 -0500 (Thu, 17 May 2007) | 1 line
  
  QPID-490 (Contributed by Carlos Medina) Implement PurgeQueue and DeleteQueue
........


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@539198 13f79535-47bb-0310-9956-ffa450edef68

22 files changed, 989 insertions, 811 deletions

diff --git a/dotnet/Qpid.Client/Client/AmqBrokerInfo.cs b/dotnet/Qpid.Client/Client/AmqBrokerInfo.cs
index 9ae1a49473..90e3788f5a 100644
--- a/dotnet/Qpid.Client/Client/AmqBrokerInfo.cs
+++ b/dotnet/Qpid.Client/Client/AmqBrokerInfo.cs
@@ -260,7 +260,7 @@ namespace Qpid.Client
             sb.Append(_transport);
             sb.Append("://");
 
-            if (!(_transport.ToLower().Equals("vm")))
+            if (!(StringEqualsIgnoreCase(_transport, "vm")))
             {
                 sb.Append(_host);
             }
@@ -268,8 +268,7 @@ namespace Qpid.Client
             sb.Append(':');
             sb.Append(_port);
 
-            // XXX
-//            sb.Append(printOptionsURL());
+            sb.Append(URLHelper.printOptions(_options));
 
             return sb.ToString();
         }
@@ -284,7 +283,8 @@ namespace Qpid.Client
 	        IBrokerInfo bd = (IBrokerInfo) obj;
 	        return StringEqualsIgnoreCase(_host, bd.Host) &&
 	        	_port == bd.Port &&
-                _transport == bd.Transport;
+                StringEqualsIgnoreCase(_transport, bd.Transport) &&
+                UseSSL == bd.UseSSL;
         }
     	
 		public override int GetHashCode()
@@ -298,35 +298,6 @@ namespace Qpid.Client
             return one.ToLower().Equals(two.ToLower());
         }
 
-//        private string printOptionsURL()
-//        {
-//            stringBuffer optionsURL = new stringBuffer();
-//
-//            optionsURL.Append('?');
-//
-//            if (!(_options.isEmpty()))
-//            {
-//
-//                for (string key : _options.keySet())
-//                {
-//                    optionsURL.Append(key);
-//
-//                    optionsURL.Append("='");
-//
-//                    optionsURL.Append(_options.get(key));
-//
-//                    optionsURL.Append("'");
-//
-//                    optionsURL.Append(URLHelper.DEFAULT_OPTION_SEPERATOR);
-//                }
-//            }
-//
-//            //remove the extra DEFAULT_OPTION_SEPERATOR or the '?' if there are no options
-//            optionsURL.deleteCharAt(optionsURL.length() - 1);
-//
-//            return optionsURL.tostring();
-//        }
-
         public bool UseSSL
         {
             get
diff --git a/dotnet/Qpid.Client/Client/AmqChannel.cs b/dotnet/Qpid.Client/Client/AmqChannel.cs
index 3471ac3640..9a8b9f787a 100644
--- a/dotnet/Qpid.Client/Client/AmqChannel.cs
+++ b/dotnet/Qpid.Client/Client/AmqChannel.cs
@@ -36,7 +36,7 @@ namespace Qpid.Client
     {
         private static readonly ILog _logger = LogManager.GetLogger(typeof(AmqChannel));
 
-        private const int BASIC_CONTENT_TYPE = 60;
+        internal const int BASIC_CONTENT_TYPE = 60;
 
         private static int _nextSessionNumber = 0;
 
@@ -122,7 +122,7 @@ namespace Qpid.Client
 
                     if (consumer == null)
                     {
-                        _logger.Warn("Received a message from queue " + message.DeliverBody.ConsumerTag + " without a handler - ignoring...");
+                        _logger.Warn("Received a message from queue " + message.DeliverBody.ConsumerTag + " without a f - ignoring...");
                     }
                     else
                     {
@@ -156,103 +156,72 @@ namespace Qpid.Client
             }
         }
 
-        internal AmqChannel(AMQConnection con, ushort channelId, bool transacted, AcknowledgeMode acknowledgeMode, int defaultPrefetch) :
-            this(con, channelId, transacted, acknowledgeMode, MessageFactoryRegistry.NewDefaultRegistry(), defaultPrefetch)
-        {
-        }
-
         /// <summary>
         /// Initializes a new instance of the <see cref="AmqChannel"/> class.
         /// </summary>
-        /// <param name="con">The con.</param>
+        /// <param name="con">The connection.</param>
         /// <param name="channelId">The channel id.</param>
         /// <param name="transacted">if set to <c>true</c> [transacted].</param>
         /// <param name="acknowledgeMode">The acknowledge mode.</param>
-        /// <param name="messageFactoryRegistry">The message factory registry.</param>
-        internal AmqChannel(AMQConnection con, ushort channelId, bool transacted, AcknowledgeMode acknowledgeMode,
-                            MessageFactoryRegistry messageFactoryRegistry, int defaultPrefetch)
+        /// <param name="defaultPrefetch">Default prefetch value</param>
+        internal AmqChannel(AMQConnection con, ushort channelId, bool transacted, AcknowledgeMode acknowledgeMode, int defaultPrefetch)
+           : this()
+        {
+           _sessionNumber = Interlocked.Increment(ref _nextSessionNumber);
+           _connection = con;
+           _transacted = transacted;
+           if ( transacted )
+           {
+              _acknowledgeMode = AcknowledgeMode.SessionTransacted;
+           } else
+           {
+              _acknowledgeMode = acknowledgeMode;
+           }
+           _channelId = channelId;
+        }
+        
+        private AmqChannel()
         {
-            _sessionNumber = Interlocked.Increment(ref _nextSessionNumber);
-            _connection = con;
-            _transacted = transacted;
-            if (transacted)
-            {
-                _acknowledgeMode = AcknowledgeMode.SessionTransacted;
-            }
-            else
-            {
-                _acknowledgeMode = acknowledgeMode;
-            }
-            _channelId = channelId;
-            _messageFactoryRegistry = messageFactoryRegistry;
+           _messageFactoryRegistry = MessageFactoryRegistry.NewDefaultRegistry();
+        }
+
+        /// <summary>
+        /// Create a disconnected channel that will fault
+        /// for most things, but is useful for testing
+        /// </summary>
+        /// <returns>A new disconnected channel</returns>
+        public static IChannel CreateDisconnectedChannel()
+        {
+           return new AmqChannel();
         }
 
+
         public IBytesMessage CreateBytesMessage()
         {
-            lock (_connection.FailoverMutex)
-            {
-                CheckNotClosed();
-                try
-                {
-                    return (IBytesMessage)_messageFactoryRegistry.CreateMessage("application/octet-stream");
-                }
-                catch (AMQException e)
-                {
-                    throw new QpidException("Unable to create message: " + e);
-                }
-            }
+            return (IBytesMessage)_messageFactoryRegistry.CreateMessage("application/octet-stream");
         }
 
         public IMessage CreateMessage()
         {
-            lock (_connection.FailoverMutex)
-            {
-                CheckNotClosed();
-                try
-                {
-                    // TODO: this is supposed to create a message consisting only of message headers
-                    return (IBytesMessage)_messageFactoryRegistry.CreateMessage("application/octet-stream");
-                }
-                catch (AMQException e)
-                {
-                    throw new QpidException("Unable to create message: " + e);
-                }
-            }
+            // TODO: this is supposed to create a message consisting only of message headers
+            return (IBytesMessage)_messageFactoryRegistry.CreateMessage("application/octet-stream");
+        }
+        
+        public IMessage CreateMessage(string mimeType)
+        {
+           return _messageFactoryRegistry.CreateMessage(mimeType);
         }
 
         public ITextMessage CreateTextMessage()
         {
-            lock (_connection.FailoverMutex)
-            {
-                CheckNotClosed();
-
-                try
-                {
-                    return (ITextMessage)_messageFactoryRegistry.CreateMessage("text/plain");
-                }
-                catch (AMQException e)
-                {
-                    throw new QpidException("Unable to create message: " + e);
-                }
-            }
+            return CreateTextMessage(String.Empty);
         }
 
         public ITextMessage CreateTextMessage(string text)
         {
-            lock (_connection.FailoverMutex)
-            {
-                CheckNotClosed();
-                try
-                {
-                    ITextMessage msg = (ITextMessage)_messageFactoryRegistry.CreateMessage("text/plain");
-                    msg.Text = text;
-                    return msg;
-                }
-                catch (AMQException e)
-                {
-                    throw new QpidException("Unable to create message: " + e);
-                }
-            }
+            ITextMessage msg = (ITextMessage)_messageFactoryRegistry.CreateMessage("text/plain");
+            msg.Text = text;
+            return msg;
         }
 
         public bool Transacted
@@ -538,11 +507,6 @@ namespace Qpid.Client
             }
         }
 
-        public IFieldTable CreateFieldTable()
-        {
-            return new FieldTable();
-        }
-
         public void Unsubscribe(String name)
         {
             throw new NotImplementedException(); // FIXME
@@ -709,6 +673,30 @@ namespace Qpid.Client
             // at this point the _consumers map will be empty
         }
 
+        public void PurgeQueue(string queueName, bool noWait)
+        {
+            DoPurgeQueue(queueName, noWait);
+        }
+
+        private void DoPurgeQueue(string queueName, bool noWait)
+        {
+            try
+            {
+                _logger.DebugFormat("PurgeQueue {0}", queueName);
+
+                AMQFrame purgeQueue = QueuePurgeBody.CreateAMQFrame(_channelId, 0, queueName, noWait);
+
+                if (noWait)
+                    _connection.ProtocolWriter.Write(purgeQueue);
+                else
+                    _connection.ConvenientProtocolWriter.SyncWrite(purgeQueue, typeof(QueuePurgeOkBody));
+            }
+            catch (AMQException)
+            {
+                throw;
+            }
+        }
+
         /**
          * Replays frame on fail over.
          * 
@@ -784,132 +772,44 @@ namespace Qpid.Client
             throw new NotImplementedException(); // FIXME
         }
 
-        public void DeleteQueue()
+        public void DeleteQueue(string queueName, bool ifUnused, bool ifEmpty, bool noWait)
         {
-            throw new NotImplementedException(); // FIXME
+            DoDeleteQueue(queueName, ifUnused, ifEmpty, noWait);
         }
 
-        public MessageConsumerBuilder CreateConsumerBuilder(string queueName)
+        private void DoDeleteQueue(string queueName, bool ifUnused, bool ifEmpty, bool noWait)
         {
-            return new MessageConsumerBuilder(this, queueName);
-        }
-
-        public MessagePublisherBuilder CreatePublisherBuilder()
-        {
-            return new MessagePublisherBuilder(this);
-        }
-
-        internal void BasicPublish(string exchangeName, string routingKey, bool mandatory, bool immediate,
-                                   AbstractQmsMessage message, DeliveryMode deliveryMode, int priority, uint timeToLive,
-                                   bool disableTimestamps)
-        {
-            DoBasicPublish(exchangeName, routingKey, mandatory, immediate, message, deliveryMode, timeToLive, priority, disableTimestamps);
-        }
-
-        private void DoBasicPublish(string exchangeName, string routingKey, bool mandatory, bool immediate, AbstractQmsMessage message, DeliveryMode deliveryMode, uint timeToLive, int priority, bool disableTimestamps)
-        {
-            AMQFrame publishFrame = BasicPublishBody.CreateAMQFrame(_channelId, 0, exchangeName,
-                                                                    routingKey, mandatory, immediate);
-
-            long currentTime = 0;
-            if (!disableTimestamps)
+            try
             {
-                currentTime = DateTime.UtcNow.Ticks;
-                message.Timestamp = currentTime;
-            }
+                _logger.Debug(string.Format("DeleteQueue name={0}", queueName));
+                
+                AMQFrame queueDelete = QueueDeleteBody.CreateAMQFrame(_channelId, 0,
+                    queueName, // queueName
+                    ifUnused, // IfUnUsed
+                    ifEmpty, // IfEmpty
+                    noWait); // NoWait
 
-            ByteBuffer buf = message.Data;
-            byte[] payload = null;
-            if (buf != null)
-            {
-                payload = new byte[buf.Remaining];
-                buf.GetBytes(payload);
-            }
-            BasicContentHeaderProperties contentHeaderProperties = message.ContentHeaderProperties;
+                _replayFrames.Add(queueDelete);
 
-            if (timeToLive > 0)
-            {
-                if (!disableTimestamps)
-                {
-                    contentHeaderProperties.Expiration = currentTime + timeToLive;
-                }
-            }
-            else
-            {
-                contentHeaderProperties.Expiration = 0;
-            }
-            contentHeaderProperties.SetDeliveryMode(deliveryMode);
-            contentHeaderProperties.Priority = (byte)priority;
-
-            ContentBody[] contentBodies = CreateContentBodies(payload);
-            AMQFrame[] frames = new AMQFrame[2 + contentBodies.Length];
-            for (int i = 0; i < contentBodies.Length; i++)
-            {
-                frames[2 + i] = ContentBody.CreateAMQFrame(_channelId, contentBodies[i]);
-            }
-            if (contentBodies.Length > 0 && _logger.IsDebugEnabled)
-            {
-                _logger.Debug(string.Format("Sending content body frames to {{exchangeName={0} routingKey={1}}}", exchangeName, routingKey));
+                if (noWait)
+                    _connection.ProtocolWriter.Write(queueDelete);
+                else
+                    _connection.ConvenientProtocolWriter.SyncWrite(queueDelete, typeof(QueueDeleteOkBody));
             }
-
-            // weight argument of zero indicates no child content headers, just bodies
-            AMQFrame contentHeaderFrame = ContentHeaderBody.CreateAMQFrame(_channelId, BASIC_CONTENT_TYPE, 0, contentHeaderProperties,
-                                                                           (uint)payload.Length);
-            if (_logger.IsDebugEnabled)
+            catch (AMQException)
             {
-                _logger.Debug(string.Format("Sending content header frame to  {{exchangeName={0} routingKey={1}}}", exchangeName, routingKey));
+                throw;
             }
+        }
 
-            frames[0] = publishFrame;
-            frames[1] = contentHeaderFrame;
-            CompositeAMQDataBlock compositeFrame = new CompositeAMQDataBlock(frames);
-
-            lock (_connection.FailoverMutex) {
-                _connection.ProtocolWriter.Write(compositeFrame);
-            }   
+        public MessageConsumerBuilder CreateConsumerBuilder(string queueName)
+        {
+            return new MessageConsumerBuilder(this, queueName);
         }
 
-        /// <summary>
-        /// Create content bodies. This will split a large message into numerous bodies depending on the negotiated
-        /// maximum frame size.
-        /// </summary>
-        /// <param name="payload"></param>
-        /// <returns>return the array of content bodies</returns>
-        private ContentBody[] CreateContentBodies(byte[] payload)
+        public MessagePublisherBuilder CreatePublisherBuilder()
         {
-            if (payload == null)
-            {
-                return null;
-            }
-            else if (payload.Length == 0)
-            {
-                return new ContentBody[0];
-            }
-            // we substract one from the total frame maximum size to account for the end of frame marker in a body frame
-            // (0xCE byte).
-            long framePayloadMax = Connection.MaximumFrameSize - 1;
-            int lastFrame = (payload.Length % framePayloadMax) > 0 ? 1 : 0;
-            int frameCount = (int)(payload.Length / framePayloadMax) + lastFrame;
-            ContentBody[] bodies = new ContentBody[frameCount];
-
-            if (frameCount == 1)
-            {
-                bodies[0] = new ContentBody();
-                bodies[0].Payload = payload;
-            }
-            else
-            {
-                long remaining = payload.Length;
-                for (int i = 0; i < bodies.Length; i++)
-                {
-                    bodies[i] = new ContentBody();
-                    byte[] framePayload = new byte[(remaining >= framePayloadMax) ? (int)framePayloadMax : (int)remaining];
-                    Array.Copy(payload, (int)framePayloadMax * i, framePayload, 0, framePayload.Length);
-                    bodies[i].Payload = framePayload;
-                    remaining -= framePayload.Length;
-                }
-            }
-            return bodies;
+            return new MessagePublisherBuilder(this);
         }
 
         public string GenerateUniqueName()
diff --git a/dotnet/Qpid.Client/Client/BasicMessageProducer.cs b/dotnet/Qpid.Client/Client/BasicMessageProducer.cs
index 759ffd62e3..fd430694df 100644
--- a/dotnet/Qpid.Client/Client/BasicMessageProducer.cs
+++ b/dotnet/Qpid.Client/Client/BasicMessageProducer.cs
@@ -21,255 +21,359 @@
 using System;
 using System.Threading;
 using log4net;
+using Qpid.Buffer;
 using Qpid.Client.Message;
 using Qpid.Messaging;
+using Qpid.Framing;
 
 namespace Qpid.Client
 {
-    public class BasicMessageProducer : Closeable, IMessagePublisher
-    {
-        protected readonly ILog _logger = LogManager.GetLogger(typeof(BasicMessageProducer));
-
-        /// <summary>
-        /// If true, messages will not get a timestamp.
-        /// </summary>
-        private bool _disableTimestamps;
-
-        /// <summary>
-        /// Priority of messages created by this producer.
-        /// </summary>
-        private int _messagePriority;
-
-        /// <summary>
-        /// Time to live of messages. Specified in milliseconds but AMQ has 1 second resolution.
-        ///
-        private long _timeToLive;
-
-        /// <summary>
-        /// Delivery mode used for this producer.
-        /// </summary>
-        private DeliveryMode _deliveryMode;
-
-        private bool _immediate;
-        private bool _mandatory;
-
-        string _exchangeName;
-        string _routingKey;
-
-        /// <summary>
-        /// Default encoding used for messages produced by this producer.
-        /// </summary>
-        private string _encoding;
-
-        /// <summary>
-        /// Default encoding used for message produced by this producer.
-        /// </summary>
-        private string _mimeType;
-
-        /// <summary>
-        /// True if this producer was created from a transacted session
-        /// </summary>
-        private bool _transacted;
-
-        private ushort _channelId;
-
-        /// <summary>
-        /// This is an id generated by the session and is used to tie individual producers to the session. This means we
-        /// can deregister a producer with the session when the producer is closed. We need to be able to tie producers
-        /// to the session so that when an error is propagated to the session it can close the producer (meaning that
-        /// a client that happens to hold onto a producer reference will get an error if he tries to use it subsequently).
-        /// </summary>
-        private long _producerId;
-
-        /// <summary>
-        /// The session used to create this producer
-        /// </summary>
-        private AmqChannel _channel;
-
-        /// <summary>
-        /// Default value for immediate flag is false, i.e. a consumer does not need to be attached to a queue
-        /// </summary>
-        protected const bool DEFAULT_IMMEDIATE = false;
-
-        /// <summary>
-        /// Default value for mandatory flag is true, i.e. server will not silently drop messages where no queue is
-        /// connected to the exchange for the message
-        /// </summary>
-        protected const bool DEFAULT_MANDATORY = true;
-
-        public BasicMessageProducer(string exchangeName, string routingKey,
-            bool transacted, 
-            ushort channelId,
-            AmqChannel channel, 
-            long producerId,
-            DeliveryMode deliveryMode,
-            long timeToLive, 
-            bool immediate, 
-            bool mandatory, 
-            int priority)
-        {
-            _exchangeName = exchangeName;
-            _routingKey = routingKey;
-            _transacted = transacted;
-            _channelId = channelId;
-            _channel = channel;
-            _producerId = producerId;
-            _deliveryMode = deliveryMode;
-            _timeToLive = timeToLive;
-            _immediate = immediate;
-            _mandatory = mandatory;
-            _messagePriority = priority;
-            
-            _channel.RegisterProducer(producerId, this);
-        }
-
-
-        #region IMessagePublisher Members
-
-        public DeliveryMode DeliveryMode
-        {
-            get
-            {
-                CheckNotClosed();
-                return _deliveryMode;
-            }
-            set
-            {
-                CheckNotClosed();
-                _deliveryMode = value;
-            }
-        }
-
-        public string ExchangeName
-        {
-            get { return _exchangeName; }
-        }
-
-        public string RoutingKey
-        {
-            get { return _routingKey; }
-        }
-
-        public bool DisableMessageID
-        {
-            get
-            {
-                throw new Exception("The method or operation is not implemented.");
-            }
-            set
-            {
-                throw new Exception("The method or operation is not implemented.");
-            }
-        }
-
-        public bool DisableMessageTimestamp
-        {
-            get
-            {
-                CheckNotClosed();
-                return _disableTimestamps;
-            }
-            set
-            {
-                CheckNotClosed();
-                _disableTimestamps = value;
-            }
-        }
-
-        public int Priority
-        {
-            get
-            {
-                CheckNotClosed();
-                return _messagePriority;
-            }
-            set
-            {
-                CheckNotClosed();
-                if (value < 0 || value > 9)
-                {
-                    throw new ArgumentOutOfRangeException("Priority of " + value + " is illegal. Value must be in range 0 to 9");
-                }
-                _messagePriority = value;
-            }
-        }
-
-        public override void Close()
-        {
-            _logger.Debug("Closing producer " + this);
-            Interlocked.Exchange(ref _closed, CLOSED);
-            _channel.DeregisterProducer(_producerId);
-        }
-
-        public void Send(IMessage msg, DeliveryMode deliveryMode, int priority, long timeToLive)
-        {
+   public class BasicMessageProducer : Closeable, IMessagePublisher
+   {
+      protected readonly ILog _logger = LogManager.GetLogger(typeof(BasicMessageProducer));
+
+      /// <summary>
+      /// If true, messages will not get a timestamp.
+      /// </summary>
+      private bool _disableTimestamps;
+
+      /// <summary>
+      /// Priority of messages created by this producer.
+      /// </summary>
+      private int _messagePriority;
+
+      /// <summary>
+      /// Time to live of messages. Specified in milliseconds but AMQ has 1 second resolution.
+      ///
+      private long _timeToLive;
+
+      /// <summary>
+      /// Delivery mode used for this producer.
+      /// </summary>
+      private DeliveryMode _deliveryMode;
+
+      private bool _immediate;
+      private bool _mandatory;
+
+      string _exchangeName;
+      string _routingKey;
+
+      /// <summary>
+      /// Default encoding used for messages produced by this producer.
+      /// </summary>
+      private string _encoding;
+
+      /// <summary>
+      /// Default encoding used for message produced by this producer.
+      /// </summary>
+      private string _mimeType;
+
+      /// <summary>
+      /// True if this producer was created from a transacted session
+      /// </summary>
+      private bool _transacted;
+
+      private ushort _channelId;
+
+      /// <summary>
+      /// This is an id generated by the session and is used to tie individual producers to the session. This means we
+      /// can deregister a producer with the session when the producer is closed. We need to be able to tie producers
+      /// to the session so that when an error is propagated to the session it can close the producer (meaning that
+      /// a client that happens to hold onto a producer reference will get an error if he tries to use it subsequently).
+      /// </summary>
+      private long _producerId;
+
+      /// <summary>
+      /// The session used to create this producer
+      /// </summary>
+      private AmqChannel _channel;
+
+      /// <summary>
+      /// Default value for immediate flag is false, i.e. a consumer does not need to be attached to a queue
+      /// </summary>
+      protected const bool DEFAULT_IMMEDIATE = false;
+
+      /// <summary>
+      /// Default value for mandatory flag is true, i.e. server will not silently drop messages where no queue is
+      /// connected to the exchange for the message
+      /// </summary>
+      protected const bool DEFAULT_MANDATORY = true;
+
+      public BasicMessageProducer(string exchangeName, string routingKey,
+          bool transacted,
+          ushort channelId,
+          AmqChannel channel,
+          long producerId,
+          DeliveryMode deliveryMode,
+          long timeToLive,
+          bool immediate,
+          bool mandatory,
+          int priority)
+      {
+         _exchangeName = exchangeName;
+         _routingKey = routingKey;
+         _transacted = transacted;
+         _channelId = channelId;
+         _channel = channel;
+         _producerId = producerId;
+         _deliveryMode = deliveryMode;
+         _timeToLive = timeToLive;
+         _immediate = immediate;
+         _mandatory = mandatory;
+         _messagePriority = priority;
+
+         _channel.RegisterProducer(producerId, this);
+      }
+
+
+      #region IMessagePublisher Members
+
+      public DeliveryMode DeliveryMode
+      {
+         get
+         {
             CheckNotClosed();
-            SendImpl(_exchangeName, _routingKey, (AbstractQmsMessage)msg, deliveryMode, priority, (uint)timeToLive, DEFAULT_MANDATORY,
-                     DEFAULT_IMMEDIATE);
-        }
-
-        public void Send(IMessage msg)
-        {
+            return _deliveryMode;
+         }
+         set
+         {
             CheckNotClosed();
-            SendImpl(_exchangeName, _routingKey, (AbstractQmsMessage)msg, _deliveryMode, _messagePriority, (uint)_timeToLive,
-                     DEFAULT_MANDATORY, DEFAULT_IMMEDIATE);
-        }
-
-        // This is a short-term hack (knowing that this code will be re-vamped sometime soon)
-        // to facilitate publishing messages to potentially non-existent recipients.
-        public void Send(IMessage msg, bool mandatory)
-        {
+            _deliveryMode = value;
+         }
+      }
+
+      public string ExchangeName
+      {
+         get { return _exchangeName; }
+      }
+
+      public string RoutingKey
+      {
+         get { return _routingKey; }
+      }
+
+      public bool DisableMessageID
+      {
+         get
+         {
+            throw new Exception("The method or operation is not implemented.");
+         }
+         set
+         {
+            throw new Exception("The method or operation is not implemented.");
+         }
+      }
+
+      public bool DisableMessageTimestamp
+      {
+         get
+         {
             CheckNotClosed();
-            SendImpl(_exchangeName, _routingKey, (AbstractQmsMessage)msg, _deliveryMode, _messagePriority, (uint)_timeToLive,
-                     mandatory, DEFAULT_IMMEDIATE);
-        }
-
-        public long TimeToLive
-        {
-            get
-            {
-                CheckNotClosed();
-                return _timeToLive;
-            }
-            set
+            return _disableTimestamps;
+         }
+         set
+         {
+            CheckNotClosed();
+            _disableTimestamps = value;
+         }
+      }
+
+      public int Priority
+      {
+         get
+         {
+            CheckNotClosed();
+            return _messagePriority;
+         }
+         set
+         {
+            CheckNotClosed();
+            if ( value < 0 || value > 9 )
             {
-                CheckNotClosed();
-                if (value < 0)
-                {
-                    throw new ArgumentOutOfRangeException("Time to live must be non-negative - supplied value was " + value);
-                }
-                _timeToLive = value;
+               throw new ArgumentOutOfRangeException("Priority of " + value + " is illegal. Value must be in range 0 to 9");
             }
-        }
-
-        #endregion
-
-        private void SendImpl(string exchangeName, string routingKey, AbstractQmsMessage message, DeliveryMode deliveryMode, int priority, uint timeToLive, bool mandatory, bool immediate)
-        {
-            _channel.BasicPublish(exchangeName, routingKey, mandatory, immediate, message, deliveryMode, priority, timeToLive, _disableTimestamps);
-        }
-
-        public string MimeType
-        {
-            set
+            _messagePriority = value;
+         }
+      }
+
+      public override void Close()
+      {
+         _logger.Debug("Closing producer " + this);
+         Interlocked.Exchange(ref _closed, CLOSED);
+         _channel.DeregisterProducer(_producerId);
+      }
+
+      public void Send(IMessage msg, DeliveryMode deliveryMode, int priority, long timeToLive)
+      {
+         CheckNotClosed();
+         SendImpl(_exchangeName, _routingKey, (AbstractQmsMessage)msg, deliveryMode, priority, (uint)timeToLive, DEFAULT_MANDATORY,
+                  DEFAULT_IMMEDIATE);
+      }
+
+      public void Send(IMessage msg)
+      {
+         CheckNotClosed();
+         SendImpl(_exchangeName, _routingKey, (AbstractQmsMessage)msg, _deliveryMode, _messagePriority, (uint)_timeToLive,
+                  DEFAULT_MANDATORY, DEFAULT_IMMEDIATE);
+      }
+
+      // This is a short-term hack (knowing that this code will be re-vamped sometime soon)
+      // to facilitate publishing messages to potentially non-existent recipients.
+      public void Send(IMessage msg, bool mandatory)
+      {
+         CheckNotClosed();
+         SendImpl(_exchangeName, _routingKey, (AbstractQmsMessage)msg, _deliveryMode, _messagePriority, (uint)_timeToLive,
+                  mandatory, DEFAULT_IMMEDIATE);
+      }
+
+      public long TimeToLive
+      {
+         get
+         {
+            CheckNotClosed();
+            return _timeToLive;
+         }
+         set
+         {
+            CheckNotClosed();
+            if ( value < 0 )
             {
-                CheckNotClosed();
-                _mimeType = value;
+               throw new ArgumentOutOfRangeException("Time to live must be non-negative - supplied value was " + value);
             }
-        }
+            _timeToLive = value;
+         }
+      }
 
-        public string Encoding
-        {
-            set
-            {
-                CheckNotClosed();
-                _encoding = value;
-            }
-        }
+      #endregion
 
-        public void Dispose()
-        {
-            Close();
-        }
-    }   
+      public string MimeType
+      {
+         set
+         {
+            CheckNotClosed();
+            _mimeType = value;
+         }
+      }
+
+      public string Encoding
+      {
+         set
+         {
+            CheckNotClosed();
+            _encoding = value;
+         }
+      }
+
+      public void Dispose()
+      {
+         Close();
+      }
+
+      #region Message Publishing
+
+      private void SendImpl(string exchangeName, string routingKey, AbstractQmsMessage message, DeliveryMode deliveryMode, int priority, uint timeToLive, bool mandatory, bool immediate)
+      {
+         // todo: handle session access ticket
+         AMQFrame publishFrame = BasicPublishBody.CreateAMQFrame(
+            _channel.ChannelId, 0, exchangeName,
+            routingKey, mandatory, immediate
+            );
+
+         // fix message properties
+         if ( !_disableTimestamps )
+         {
+            message.Timestamp = DateTime.UtcNow.Ticks;
+            message.Expiration = message.Timestamp + timeToLive;
+         } else
+         {
+            message.Expiration = 0;
+         }
+         message.DeliveryMode = deliveryMode;
+         message.Priority = (byte)priority;
+
+         ByteBuffer payload = message.Data;
+         int payloadLength = payload.Limit;
+
+         ContentBody[] contentBodies = CreateContentBodies(payload);
+         AMQFrame[] frames = new AMQFrame[2 + contentBodies.Length];
+         for ( int i = 0; i < contentBodies.Length; i++ )
+         {
+            frames[2 + i] = ContentBody.CreateAMQFrame(_channelId, contentBodies[i]);
+         }
+         if ( contentBodies.Length > 0 && _logger.IsDebugEnabled )
+         {
+            _logger.Debug(string.Format("Sending content body frames to {{exchangeName={0} routingKey={1}}}", exchangeName, routingKey));
+         }
+
+         // weight argument of zero indicates no child content headers, just bodies
+         AMQFrame contentHeaderFrame = ContentHeaderBody.CreateAMQFrame(
+            _channelId, AmqChannel.BASIC_CONTENT_TYPE, 0, 
+            message.ContentHeaderProperties, (uint)payloadLength
+            );
+         if ( _logger.IsDebugEnabled )
+         {
+            _logger.Debug(string.Format("Sending content header frame to  {{exchangeName={0} routingKey={1}}}", exchangeName, routingKey));
+         }
+
+         frames[0] = publishFrame;
+         frames[1] = contentHeaderFrame;
+         CompositeAMQDataBlock compositeFrame = new CompositeAMQDataBlock(frames);
+
+         lock ( _channel.Connection.FailoverMutex )
+         {
+            _channel.Connection.ProtocolWriter.Write(compositeFrame);
+         }
+      }
+
+
+      /// <summary>
+      /// Create content bodies. This will split a large message into numerous bodies depending on the negotiated
+      /// maximum frame size.
+      /// </summary>
+      /// <param name="payload"></param>
+      /// <returns>return the array of content bodies</returns>
+      private ContentBody[] CreateContentBodies(ByteBuffer payload)
+      {
+         if ( payload == null )
+         {
+            return null;
+         } else if ( payload.Remaining == 0 )
+         {
+            return new ContentBody[0];
+         }
+         // we substract one from the total frame maximum size to account for the end of frame marker in a body frame
+         // (0xCE byte).
+         int framePayloadMax = (int)(_channel.Connection.MaximumFrameSize - 1);
+         int frameCount = CalculateContentBodyFrames(payload);
+         ContentBody[] bodies = new ContentBody[frameCount];
+         for ( int i = 0; i < frameCount; i++ )
+         {
+            int length = (payload.Remaining >= framePayloadMax)
+               ? framePayloadMax : payload.Remaining;
+            bodies[i] = new ContentBody(payload, (uint)length);
+         }
+         return bodies;
+      }
+
+      private int CalculateContentBodyFrames(ByteBuffer payload)
+      {
+         // we substract one from the total frame maximum size to account 
+         // for the end of frame marker in a body frame
+         // (0xCE byte).
+         int frameCount;
+         if ( (payload == null) || (payload.Remaining == 0) )
+         {
+            frameCount = 0;
+         } else
+         {
+            int dataLength = payload.Remaining;
+            int framePayloadMax = (int)_channel.Connection.MaximumFrameSize - 1;
+            int lastFrame = ((dataLength % framePayloadMax) > 0) ? 1 : 0;
+            frameCount = (int)(dataLength / framePayloadMax) + lastFrame;
+         }
+
+         return frameCount;
+      }
+      #endregion // Message Publishing
+   }
 }
diff --git a/dotnet/Qpid.Client/Client/Handler/ConnectionTuneMethodHandler.cs b/dotnet/Qpid.Client/Client/Handler/ConnectionTuneMethodHandler.cs
index 41e9d2240c..afcbd26781 100644
--- a/dotnet/Qpid.Client/Client/Handler/ConnectionTuneMethodHandler.cs
+++ b/dotnet/Qpid.Client/Client/Handler/ConnectionTuneMethodHandler.cs
@@ -45,14 +45,12 @@ namespace Qpid.Client.Handler
             _logger.Debug(String.Format("ConnectionTune.heartbeat = {0}.", frame.Heartbeat));
 
             parameters.FrameMax = frame.FrameMax;
-            parameters.FrameMax = 65535;
-            //params.setChannelMax(frame.channelMax);
             parameters.Heartbeat = frame.Heartbeat;
             session.ConnectionTuneParameters = parameters;
 
             stateManager.ChangeState(AMQState.CONNECTION_NOT_OPENED);
             session.WriteFrame(ConnectionTuneOkBody.CreateAMQFrame(
-                                   evt.ChannelId, frame.ChannelMax, 65535, frame.Heartbeat));
+                                   evt.ChannelId, frame.ChannelMax, frame.FrameMax, frame.Heartbeat));
             session.WriteFrame(ConnectionOpenBody.CreateAMQFrame(
                                    evt.ChannelId, session.AMQConnection.VirtualHost, null, true));
 
diff --git a/dotnet/Qpid.Client/Client/Handler/QueueDeleteOkMethodHandler.cs b/dotnet/Qpid.Client/Client/Handler/QueueDeleteOkMethodHandler.cs
new file mode 100644
index 0000000000..9a67b69834
--- /dev/null
+++ b/dotnet/Qpid.Client/Client/Handler/QueueDeleteOkMethodHandler.cs
@@ -0,0 +1,44 @@
+/*

+ *

+ * 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 log4net;

+using Qpid.Client.Message;

+using Qpid.Client.Protocol;

+using Qpid.Client.State;

+using Qpid.Framing;

+

+namespace Qpid.Client.Handler

+{

+    public class QueueDeleteOkMethodHandler : IStateAwareMethodListener

+    {

+

+        private static readonly ILog _logger = LogManager.GetLogger(typeof(QueueDeleteOkMethodHandler));

+

+        public void MethodReceived(AMQStateManager stateManager, AMQMethodEvent evt)

+        {

+            QueueDeleteOkBody body = (QueueDeleteOkBody)evt.Method;

+            if (body != null)

+            {

+                _logger.InfoFormat("Received Queue.Delete-Ok message, message count {0}", body.MessageCount);

+            }

+        }

+

+    }

+}

diff --git a/dotnet/Qpid.Client/Client/Handler/QueuePurgeOkMethodHandler.cs b/dotnet/Qpid.Client/Client/Handler/QueuePurgeOkMethodHandler.cs
new file mode 100644
index 0000000000..59e7db34ab
--- /dev/null
+++ b/dotnet/Qpid.Client/Client/Handler/QueuePurgeOkMethodHandler.cs
@@ -0,0 +1,44 @@
+/*

+ *

+ * 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 log4net;

+using Qpid.Client.Message;

+using Qpid.Client.Protocol;

+using Qpid.Client.State;

+using Qpid.Framing;

+

+namespace Qpid.Client.Handler

+{

+    public class QueuePurgeOkMethodHandler : IStateAwareMethodListener

+    {

+

+        private static readonly ILog _logger = LogManager.GetLogger(typeof(QueuePurgeOkMethodHandler));

+

+        public void MethodReceived(AMQStateManager stateManager, AMQMethodEvent evt)

+        {

+            QueuePurgeOkBody body = (QueuePurgeOkBody)evt.Method;

+            if (body != null)

+            {

+                _logger.InfoFormat("Received Queue.Purge-Ok message, message count {0}", body.MessageCount);

+            }

+        }

+

+    }

+}

diff --git a/dotnet/Qpid.Client/Client/Message/AMQMessageFactory.cs b/dotnet/Qpid.Client/Client/Message/AMQMessageFactory.cs
index 75c4edd67d..a7ee085a04 100644
--- a/dotnet/Qpid.Client/Client/Message/AMQMessageFactory.cs
+++ b/dotnet/Qpid.Client/Client/Message/AMQMessageFactory.cs
@@ -27,7 +27,7 @@ namespace Qpid.Client.Message
 {
     public abstract class AbstractQmsMessageFactory : IMessageFactory
     {
-        public abstract AbstractQmsMessage CreateMessage();
+        public abstract AbstractQmsMessage CreateMessage(string mimeType);
 
         private static readonly ILog _logger = LogManager.GetLogger(typeof (AbstractQmsMessageFactory));
 
@@ -43,7 +43,7 @@ namespace Qpid.Client.Message
             if (bodies != null && bodies.Count == 1)
             {
                 _logger.Debug("Non-fragmented message body (bodySize=" + contentHeader.BodySize +")");
-                data = ByteBuffer.Wrap(((ContentBody)bodies[0]).Payload);
+                data = ((ContentBody)bodies[0]).Payload;
             }
             else
             {
diff --git a/dotnet/Qpid.Client/Client/Message/AbstractQmsMessage.cs b/dotnet/Qpid.Client/Client/Message/AbstractQmsMessage.cs
index 1d2b2db3ca..8e90e852dd 100644
--- a/dotnet/Qpid.Client/Client/Message/AbstractQmsMessage.cs
+++ b/dotnet/Qpid.Client/Client/Message/AbstractQmsMessage.cs
@@ -72,24 +72,29 @@ namespace Qpid.Client.Message
 
 #endregion
 
+        #region Properties
+        //
+        // Properties
+        //
+
+        /// <summary>
+        /// The application message identifier
+        /// </summary>
         public string MessageId
         {
-            get
-            {
+            get {
                 if (ContentHeaderProperties.MessageId == null)
                 {
                     ContentHeaderProperties.MessageId = "ID:" + DeliveryTag;
                 }
                 return ContentHeaderProperties.MessageId;
             }
-            set
-            {
-                ContentHeaderProperties.MessageId = value;
-            }
-
-
-        }        
+            set { ContentHeaderProperties.MessageId = value; }
+        }
 
+        /// <summary>
+        /// The message timestamp
+        /// </summary>
         public long Timestamp
         {
             get
@@ -103,36 +108,22 @@ namespace Qpid.Client.Message
             }
         }        
 
-        protected void CheckReadable()
-        {
-            if (!_readableMessage)
-            {
-                throw new MessageNotReadableException("You need to call reset() to make the message readable");
-            }
-        }
-
+        /// <summary>
+        /// The <see cref="CorrelationId"/> as a byte array.
+        /// </summary>
         public byte[] CorrelationIdAsBytes
         {
-            get
-            {            
-                return Encoding.Default.GetBytes(ContentHeaderProperties.CorrelationId);
-            }
-            set
-            {
-                ContentHeaderProperties.CorrelationId = Encoding.Default.GetString(value);
-            }
+            get { return Encoding.Default.GetBytes(ContentHeaderProperties.CorrelationId); }
+            set { ContentHeaderProperties.CorrelationId = Encoding.Default.GetString(value); }
         }
 
+        /// <summary>
+        /// The application correlation identifier
+        /// </summary>
         public string CorrelationId
         {
-            get
-            {
-                return ContentHeaderProperties.CorrelationId;
-            }
-            set
-            {
-                ContentHeaderProperties.ContentType = value;
-            }
+            get { return ContentHeaderProperties.CorrelationId; }
+            set { ContentHeaderProperties.CorrelationId = value; }
         }
         
         struct Dest
@@ -147,6 +138,9 @@ namespace Qpid.Client.Message
             }
         }
 
+        /// <summary>
+        /// Exchange name of the reply-to address
+        /// </summary>
         public string ReplyToExchangeName
         {
             get
@@ -162,6 +156,9 @@ namespace Qpid.Client.Message
             }
         }
 
+        /// <summary>
+        /// Routing key of the reply-to address
+        /// </summary>
         public string ReplyToRoutingKey
         {
             get
@@ -177,50 +174,11 @@ namespace Qpid.Client.Message
             }
         }
 
-        /// <summary>
-        /// Decodes the replyto field if one is set.
-        /// 
-        /// Splits a replyto field containing an exchange name followed by a ':', followed by a routing key into the exchange name and
-        /// routing key seperately. The exchange name may be empty in which case the empty string is returned. If the exchange name is
-        /// empty the replyto field is expected to being with ':'.
-        /// 
-        /// Anyhting other than a two part replyto field sperated with a ':' will result in an exception.
-        /// </summary>
-        /// 
-        /// <returns>A destination initialized to the replyto location if a replyto field was set, or an empty destination otherwise.</returns>
-        private Dest ReadReplyToHeader()
-        {
-            string replyToEncoding = ContentHeaderProperties.ReplyTo;
-
-            if (replyToEncoding == null)
-            {
-                return new Dest();
-            }
-            else
-            {
-                // Split the replyto field on a ':'
-                string[] split = replyToEncoding.Split(':');
 
-                // Ensure that the replyto field argument only consisted of two parts.
-                if (split.Length != 2)
-                {
-                    throw new QpidException("Illegal value in ReplyTo property: " + replyToEncoding);
-                }
-
-                // Extract the exchange name and routing key from the split replyto field.
-                string exchangeName = split[0];
-                string routingKey = split[1];
-
-                return new Dest(exchangeName, routingKey);                
-            }            
-        }
-        
-        private void WriteReplyToHeader(Dest dest)
-        {
-            string encodedDestination = string.Format("{0}:{1}", dest.ExchangeName, dest.RoutingKey);
-            ContentHeaderProperties.ReplyTo = encodedDestination;            
-        }
 
+        /// <summary>
+        /// Non-persistent (1) or persistent (2)
+        /// </summary>
         public DeliveryMode DeliveryMode
         {
             get
@@ -242,100 +200,94 @@ namespace Qpid.Client.Message
             }
         }        
 
+       /// <summary>
+       /// True, if this is a redelivered message
+       /// </summary>
         public bool Redelivered
         {
-            get
-            {
-                return _redelivered;
-            }
-            set
-            {
-                _redelivered = value;
-            }
-        }        
+            get { return _redelivered; }
+            set { _redelivered = value; }
+        }
 
+        /// <summary>
+        /// The message type name
+        /// </summary>
         public string Type
         {
-            get
-            {
-                return MimeType;
-            }            
-            set
-            {
-                //MimeType = value;
-            }
+            get { return ContentHeaderProperties.Type; }
+            set { ContentHeaderProperties.Type = value; }
         }
-        
+
+        /// <summary>
+        /// Message expiration specification
+        /// </summary>
         public long Expiration
         {
-            get
-            {
-                return ContentHeaderProperties.Expiration;
-            }
-            set
-            {
-                ContentHeaderProperties.Expiration = value;
-            }
+            get { return ContentHeaderProperties.Expiration; }
+            set { ContentHeaderProperties.Expiration = value; }
         }
 
-        public int Priority
+        /// <summary>
+        /// The message priority, 0 to 9
+        /// </summary>
+        public byte Priority
         {
-            get
-            {
-                return ContentHeaderProperties.Priority;
-            }
-            set
-            {
-                ContentHeaderProperties.Priority = (byte) value;
-            }
+            get { return ContentHeaderProperties.Priority; }
+            set { ContentHeaderProperties.Priority = (byte) value; }
         }
 
-        // FIXME: implement
+        /// <summary>
+        /// The MIME Content Type
+        /// </summary>
         public string ContentType
         {
-            get { throw new NotImplementedException(); }
-            set { throw new NotImplementedException(); }
+            get { return ContentHeaderProperties.ContentType; }
+            set { ContentHeaderProperties.ContentType = value; }
         }
 
-        // FIXME: implement
+        /// <summary>
+        /// The MIME Content Encoding
+        /// </summary>
         public string ContentEncoding
         {
-            get { throw new NotImplementedException(); }
-            set { throw new NotImplementedException(); }
-        }
-
-        public void Acknowledge()
-        {
-            // the JMS 1.1 spec says in section 3.6 that calls to acknowledge are ignored when client acknowledge
-            // is not specified. In our case, we only set the session field where client acknowledge mode is specified.
-            if (_channel != null)
-            {
-                // we set multiple to true here since acknowledgement implies acknowledge of all previous messages
-                // received on the session
-                _channel.AcknowledgeMessage((ulong)DeliveryTag, true);
-            }
-
+            get { return ContentHeaderProperties.Encoding; }
+            set { ContentHeaderProperties.Encoding = value; }
         }
 
+        /// <summary>
+        /// Headers of this message
+        /// </summary>
         public IHeaders Headers
         {
             get { return _headers; }
         }
 
-        public abstract void ClearBodyImpl();
+        /// <summary>
+        /// The creating user id
+        /// </summary>
+        public string UserId
+        {
+            get { return ContentHeaderProperties.UserId; }
+            set { ContentHeaderProperties.UserId = value; }
+        }
 
-        public void ClearBody()
+        /// <summary>
+        /// The creating application id
+        /// </summary>
+        public string AppId
         {
-            ClearBodyImpl();
-            _readableMessage = false;
+            get { return ContentHeaderProperties.AppId; }
+            set { ContentHeaderProperties.AppId = value; }
         }
 
         /// <summary>
-        /// Get a String representation of the body of the message. Used in the
-        /// toString() method which outputs this before message properties.
+        /// Intra-cluster routing identifier
         /// </summary>
-        /// <exception cref="QpidException"></exception>
-        public abstract string ToBodyString();
+        public string ClusterId
+        {
+            get { return ContentHeaderProperties.ClusterId; }
+            set { ContentHeaderProperties.ClusterId = value; }
+        }
 
         /// <summary>
         /// Return the raw byte array that is used to populate the frame when sending
@@ -367,12 +319,37 @@ namespace Qpid.Client.Message
                 _data = value;
             }
         }
+        #endregion // Properties
+
 
-        public abstract string MimeType
+        public void Acknowledge()
         {
-            get;           
+            // the JMS 1.1 spec says in section 3.6 that calls to acknowledge are ignored when client acknowledge
+            // is not specified. In our case, we only set the session field where client acknowledge mode is specified.
+            if (_channel != null)
+            {
+                // we set multiple to true here since acknowledgement implies acknowledge of all previous messages
+                // received on the session
+                _channel.AcknowledgeMessage((ulong)DeliveryTag, true);
+            }
+
         }
 
+        public abstract void ClearBodyImpl();
+
+        public void ClearBody()
+        {
+            ClearBodyImpl();
+            _readableMessage = false;
+        }
+
+        /// <summary>
+        /// Get a String representation of the body of the message. Used in the
+        /// toString() method which outputs this before message properties.
+        /// </summary>
+        /// <exception cref="QpidException"></exception>
+        public abstract string ToBodyString();
+
         public override string ToString()
         {
             try
@@ -403,18 +380,6 @@ namespace Qpid.Client.Message
             }
         }
 
-        public IFieldTable UnderlyingMessagePropertiesMap
-        {
-            get
-            {
-                return ContentHeaderProperties.Headers;
-            }
-            set
-            {
-                ContentHeaderProperties.Headers = (FieldTable)value;
-            }
-        }
-        
         public FieldTable PopulateHeadersFromMessageProperties()
         {
             if (ContentHeaderProperties.Headers == null)
@@ -466,5 +431,56 @@ namespace Qpid.Client.Message
         {
             get { return !_readableMessage; }
         }
+
+        protected void CheckReadable()
+        {
+           if ( !_readableMessage )
+           {
+              throw new MessageNotReadableException("You need to call reset() to make the message readable");
+           }
+        }
+
+        /// <summary>
+        /// Decodes the replyto field if one is set.
+        /// 
+        /// Splits a replyto field containing an exchange name followed by a ':', followed by a routing key into the exchange name and
+        /// routing key seperately. The exchange name may be empty in which case the empty string is returned. If the exchange name is
+        /// empty the replyto field is expected to being with ':'.
+        /// 
+        /// Anyhting other than a two part replyto field sperated with a ':' will result in an exception.
+        /// </summary>
+        /// 
+        /// <returns>A destination initialized to the replyto location if a replyto field was set, or an empty destination otherwise.</returns>
+        private Dest ReadReplyToHeader()
+        {
+           string replyToEncoding = ContentHeaderProperties.ReplyTo;
+
+           if ( replyToEncoding == null )
+           {
+              return new Dest();
+           } else
+           {
+              // Split the replyto field on a ':'
+              string[] split = replyToEncoding.Split(':');
+
+              // Ensure that the replyto field argument only consisted of two parts.
+              if ( split.Length != 2 )
+              {
+                 throw new QpidException("Illegal value in ReplyTo property: " + replyToEncoding);
+              }
+
+              // Extract the exchange name and routing key from the split replyto field.
+              string exchangeName = split[0];
+              string routingKey = split[1];
+
+              return new Dest(exchangeName, routingKey);
+           }
+        }
+
+        private void WriteReplyToHeader(Dest dest)
+        {
+           string encodedDestination = string.Format("{0}:{1}", dest.ExchangeName, dest.RoutingKey);
+           ContentHeaderProperties.ReplyTo = encodedDestination;
+        }
     }
 }
diff --git a/dotnet/Qpid.Client/Client/Message/IMessageFactory.cs b/dotnet/Qpid.Client/Client/Message/IMessageFactory.cs
index 4a109b128e..cffc585067 100644
--- a/dotnet/Qpid.Client/Client/Message/IMessageFactory.cs
+++ b/dotnet/Qpid.Client/Client/Message/IMessageFactory.cs
@@ -28,11 +28,12 @@ namespace Qpid.Client.Message
         /// <summary>
         /// Create a message
         /// </summary>
-        /// <param name="messageNbr"></param>
-        /// <param name="redelivered"></param>
-        /// <param name="contentHeader"></param>
-        /// <param name="bodies"></param>
-        /// <returns></returns>
+        /// <param name="deliverTag">Delivery Tag</param>
+        /// <param name="messageNbr">Message Sequence Number</param>
+        /// <param name="redelivered">True if this is a redelivered message</param>
+        /// <param name="contentHeader">Content headers</param>
+        /// <param name="bodies">Message bodies</param>
+        /// <returns>The new message</returns>
         /// <exception cref="QpidMessagingException">if the message cannot be created</exception>
         AbstractQmsMessage CreateMessage(long deliverTag, bool redelivered,
                                          ContentHeaderBody contentHeader,
@@ -41,9 +42,10 @@ namespace Qpid.Client.Message
         /// <summary>
         /// Creates the message.
         /// </summary>
-        /// <returns></returns>
+        /// <param name="mimeType">Mime type to associate the new message with</param>
+        /// <returns>The new message</returns>
         /// <exception cref="QpidMessagingException">if the message cannot be created</exception>
-        AbstractQmsMessage CreateMessage();
+        AbstractQmsMessage CreateMessage(string mimeType);
     }
 }
 
diff --git a/dotnet/Qpid.Client/Client/Message/MessageFactoryRegistry.cs b/dotnet/Qpid.Client/Client/Message/MessageFactoryRegistry.cs
index 95257cef8a..f854a541fc 100644
--- a/dotnet/Qpid.Client/Client/Message/MessageFactoryRegistry.cs
+++ b/dotnet/Qpid.Client/Client/Message/MessageFactoryRegistry.cs
@@ -25,93 +25,104 @@ using Qpid.Messaging;
 
 namespace Qpid.Client.Message
 {
-    public class MessageFactoryRegistry
-    {
-        private readonly Hashtable _mimeToFactoryMap = new Hashtable();
+   public class MessageFactoryRegistry
+   {
+      private readonly Hashtable _mimeToFactoryMap = new Hashtable();
+      private IMessageFactory _defaultFactory;
 
-        public void RegisterFactory(string mimeType, IMessageFactory mf)
-        {    
-            if (mf == null)
-            {
-                throw new ArgumentNullException("Message factory");
-            }
-            if (mimeType == null)
-            {
-                throw new ArgumentNullException("mf");
-            }
-            _mimeToFactoryMap[mimeType] = mf;
-        }
+      /// <summary>
+      /// Default factory to use for unknown message types
+      /// </summary>
+      public IMessageFactory DefaultFactory
+      {
+         get { return _defaultFactory; }
+         set { _defaultFactory = value; }
+      }
 
-        public void DeregisterFactory(string mimeType)
-        {
-            _mimeToFactoryMap.Remove(mimeType);
-        }
+      /// <summary>
+      /// Register a new message factory for a MIME type
+      /// </summary>
+      /// <param name="mimeType">Mime type to register</param>
+      /// <param name="mf"></param>
+      public void RegisterFactory(string mimeType, IMessageFactory mf)
+      {
+         if ( mf == null )
+            throw new ArgumentNullException("mf");
+         if ( mimeType == null || mimeType.Length == 0 )
+            throw new ArgumentNullException("mimeType");
 
-        /// <summary>
-        /// Create a message. This looks up the MIME type from the content header and instantiates the appropriate
-        /// concrete message type.
-        /// </summary>
-        /// <param name="messageNbr">the AMQ message id</param>
-        /// <param name="redelivered">true if redelivered</param>
-        /// <param name="contentHeader">the content header that was received</param>
-        /// <param name="bodies">a list of ContentBody instances</param>
-        /// <returns>the message.</returns>
-        /// <exception cref="AMQException"/>
-        /// <exception cref="QpidException"/>
-        public AbstractQmsMessage CreateMessage(long messageNbr, bool redelivered,
-                                                ContentHeaderBody contentHeader,
-                                                IList bodies)
-        {
-            BasicContentHeaderProperties properties =  (BasicContentHeaderProperties) contentHeader.Properties;
+         _mimeToFactoryMap[mimeType] = mf;
+      }
 
-            if (properties.ContentType == null)
-            {
-                properties.ContentType = "";
-            }
+      /// <summary>
+      /// Remove a message factory
+      /// </summary>
+      /// <param name="mimeType">MIME type to unregister</param>
+      public void DeregisterFactory(string mimeType)
+      {
+         _mimeToFactoryMap.Remove(mimeType);
+      }
 
-            IMessageFactory mf = (IMessageFactory) _mimeToFactoryMap[properties.ContentType];
-            if (mf == null)
-            {
-                throw new AMQException("Unsupport MIME type of " + properties.ContentType);
-            }
-            else
-            {
-                return mf.CreateMessage(messageNbr, redelivered, contentHeader, bodies);
-            }
-        }
+      /// <summary>
+      /// Create a message. This looks up the MIME type from the content header and instantiates the appropriate
+      /// concrete message type.
+      /// </summary>
+      /// <param name="messageNbr">the AMQ message id</param>
+      /// <param name="redelivered">true if redelivered</param>
+      /// <param name="contentHeader">the content header that was received</param>
+      /// <param name="bodies">a list of ContentBody instances</param>
+      /// <returns>the message.</returns>
+      /// <exception cref="AMQException"/>
+      /// <exception cref="QpidException"/>
+      public AbstractQmsMessage CreateMessage(long messageNbr, bool redelivered,
+                                              ContentHeaderBody contentHeader,
+                                              IList bodies)
+      {
+         BasicContentHeaderProperties properties = (BasicContentHeaderProperties)contentHeader.Properties;
 
-        public AbstractQmsMessage CreateMessage(string mimeType)
-        {
-            if (mimeType == null)
-            {
-                throw new ArgumentNullException("Mime type must not be null");
-            }
-            IMessageFactory mf = (IMessageFactory) _mimeToFactoryMap[mimeType];
-            if (mf == null)
-            {
-                throw new AMQException("Unsupport MIME type of " + mimeType);
-            }
-            else
-            {
-                return mf.CreateMessage();
-            }
-        }
+         if ( properties.ContentType == null )
+         {
+            properties.ContentType = "";
+         }
 
-        /// <summary>
-        /// Construct a new registry with the default message factories registered
-        /// </summary>
-        /// <returns>a message factory registry</returns>
-        public static MessageFactoryRegistry NewDefaultRegistry()
-        {
-            MessageFactoryRegistry mf = new MessageFactoryRegistry();
-            mf.RegisterFactory("text/plain", new QpidTextMessageFactory());
-            mf.RegisterFactory("text/xml", new QpidTextMessageFactory());
-            mf.RegisterFactory("application/octet-stream", new QpidBytesMessageFactory());
-            // TODO: use bytes message for default message factory            
-            // MJA - just added this bit back in...
-            mf.RegisterFactory("", new QpidBytesMessageFactory());
-            return mf;
-        }
-    }
+         IMessageFactory mf = GetFactory(properties.ContentType);
+         return mf.CreateMessage(messageNbr, redelivered, contentHeader, bodies);
+      }
+
+      /// <summary>
+      /// Create a new message of the specified type
+      /// </summary>
+      /// <param name="mimeType">The Mime type</param>
+      /// <returns>The new message</returns>
+      public AbstractQmsMessage CreateMessage(string mimeType)
+      {
+         if ( mimeType == null || mimeType.Length == 0 )
+            throw new ArgumentNullException("mimeType");
+
+         IMessageFactory mf = GetFactory(mimeType);
+         return mf.CreateMessage(mimeType);
+      }
+
+      /// <summary>
+      /// Construct a new registry with the default message factories registered
+      /// </summary>
+      /// <returns>a message factory registry</returns>
+      public static MessageFactoryRegistry NewDefaultRegistry()
+      {
+         MessageFactoryRegistry mf = new MessageFactoryRegistry();
+         mf.RegisterFactory("text/plain", new QpidTextMessageFactory());
+         mf.RegisterFactory("text/xml", new QpidTextMessageFactory());
+         mf.RegisterFactory("application/octet-stream", new QpidBytesMessageFactory());
+
+         mf.DefaultFactory = new QpidBytesMessageFactory();
+         return mf;
+      }
+
+      private IMessageFactory GetFactory(string mimeType)
+      {
+         IMessageFactory mf = (IMessageFactory)_mimeToFactoryMap[mimeType];
+         return mf != null ? mf : _defaultFactory;
+      }
+   }
 }
 
diff --git a/dotnet/Qpid.Client/Client/Message/QpidBytesMessage.cs b/dotnet/Qpid.Client/Client/Message/QpidBytesMessage.cs
index 32e47d852a..cb504d1378 100644
--- a/dotnet/Qpid.Client/Client/Message/QpidBytesMessage.cs
+++ b/dotnet/Qpid.Client/Client/Message/QpidBytesMessage.cs
@@ -43,8 +43,6 @@ namespace Qpid.Client.Message
 
     public class QpidBytesMessage : AbstractQmsMessage, IBytesMessage
     {
-        private const string MIME_TYPE = "application/octet-stream";
-
         private const int DEFAULT_BUFFER_INITIAL_SIZE = 1024;
 
         public QpidBytesMessage() : this(null)
@@ -59,7 +57,6 @@ namespace Qpid.Client.Message
         QpidBytesMessage(ByteBuffer data) : base(data)
         {
             // superclass constructor has instantiated a content header at this point
-            ContentHeaderProperties.ContentType = MIME_TYPE;
             if (data == null)
             {
                 _data = ByteBuffer.Allocate(DEFAULT_BUFFER_INITIAL_SIZE);
@@ -71,7 +68,6 @@ namespace Qpid.Client.Message
             // TODO: this casting is ugly. Need to review whole ContentHeaderBody idea
             : base(messageNbr, (BasicContentHeaderProperties)contentHeader.Properties, data)
         {
-            ContentHeaderProperties.ContentType = MIME_TYPE;        
         }
 
         public override void ClearBodyImpl()
@@ -116,14 +112,6 @@ namespace Qpid.Client.Message
             }
         }
 
-        public override string MimeType
-        {
-            get
-            {
-                return MIME_TYPE;
-            }
-        }
-
         public long BodyLength
         {
             get
diff --git a/dotnet/Qpid.Client/Client/Message/QpidBytesMessageFactory.cs b/dotnet/Qpid.Client/Client/Message/QpidBytesMessageFactory.cs
index de4c6675c7..e96c38cbac 100644
--- a/dotnet/Qpid.Client/Client/Message/QpidBytesMessageFactory.cs
+++ b/dotnet/Qpid.Client/Client/Message/QpidBytesMessageFactory.cs
@@ -62,9 +62,11 @@ namespace Qpid.Client.Message
             return new QpidBytesMessage(deliveryTag, contentHeader, data);
         }
 
-        public override AbstractQmsMessage CreateMessage()
+        public override AbstractQmsMessage CreateMessage(string mimeType)
         {
-            return new QpidBytesMessage();
+            QpidBytesMessage msg = new QpidBytesMessage();
+            msg.ContentType = mimeType;
+            return msg;
         }
 
     }
diff --git a/dotnet/Qpid.Client/Client/Message/QpidHeaders.cs b/dotnet/Qpid.Client/Client/Message/QpidHeaders.cs
index 6538fcbefc..a258c82d15 100644
--- a/dotnet/Qpid.Client/Client/Message/QpidHeaders.cs
+++ b/dotnet/Qpid.Client/Client/Message/QpidHeaders.cs
@@ -28,16 +28,10 @@ namespace Qpid.Client.Message
          _headers.Clear();
       }
 
-      public string this[string name]
+      public object this[string name]
       {
-         get
-         {
-            return GetString(name);
-         }
-         set
-         {
-            SetString(name, value);
-         }
+         get { return GetObject(name); }
+         set { SetObject(name, value); }
       }
 
       public bool GetBoolean(string name)
@@ -167,6 +161,18 @@ namespace Qpid.Client.Message
          _headers.SetString(propertyName, value);
       }
 
+      public object GetObject(string propertyName)
+      {
+         CheckPropertyName(propertyName);
+         return _headers[propertyName];
+      }
+
+      public void SetObject(string propertyName, object value)
+      {
+         CheckPropertyName(propertyName);
+         _headers[propertyName] = value;
+      }
+
       private static void CheckPropertyName(string propertyName)
       {
          if ( propertyName == null )
diff --git a/dotnet/Qpid.Client/Client/Message/QpidTextMessage.cs b/dotnet/Qpid.Client/Client/Message/QpidTextMessage.cs
index cff42f1df5..ae8bdb2074 100644
--- a/dotnet/Qpid.Client/Client/Message/QpidTextMessage.cs
+++ b/dotnet/Qpid.Client/Client/Message/QpidTextMessage.cs
@@ -28,25 +28,22 @@ namespace Qpid.Client.Message
 {
     public class QpidTextMessage : AbstractQmsMessage, ITextMessage
     {
-        private const string MIME_TYPE = "text/plain";
-
         private string _decodedValue = null;
+        private static Encoding DEFAULT_ENCODING = Encoding.UTF8;
 
         internal QpidTextMessage() : this(null, null)
         {
+           ContentEncoding = DEFAULT_ENCODING.BodyName;
         }
 
-        QpidTextMessage(ByteBuffer data, String encoding) : base(data)
+        internal QpidTextMessage(ByteBuffer data, String encoding) : base(data)
         {
-            ContentHeaderProperties.ContentType = MIME_TYPE;
-            ContentHeaderProperties.Encoding = encoding;
+            ContentEncoding = encoding;
         }
 
         internal QpidTextMessage(long deliveryTag, BasicContentHeaderProperties contentHeader, ByteBuffer data)
             :base(deliveryTag, contentHeader, data)
         {
-            contentHeader.ContentType = MIME_TYPE;
-            _data = data; // FIXME: Unnecessary - done in base class ctor.
         }
 
         public override void ClearBodyImpl()
@@ -64,14 +61,6 @@ namespace Qpid.Client.Message
             return Text;
         }
 
-        public override string MimeType
-        {
-            get
-            {
-                return MIME_TYPE;
-            }
-        }        
-
         public string Text
         {
             get
@@ -100,7 +89,7 @@ namespace Qpid.Client.Message
                     }
                     else
                     {
-                        _decodedValue = Encoding.Default.GetString(bytes);
+                        _decodedValue = DEFAULT_ENCODING.GetString(bytes);
                     }
                     return _decodedValue;                    
                 }
diff --git a/dotnet/Qpid.Client/Client/Message/QpidTextMessageFactory.cs b/dotnet/Qpid.Client/Client/Message/QpidTextMessageFactory.cs
index cc4f6dafe1..4730fa56ad 100644
--- a/dotnet/Qpid.Client/Client/Message/QpidTextMessageFactory.cs
+++ b/dotnet/Qpid.Client/Client/Message/QpidTextMessageFactory.cs
@@ -25,9 +25,11 @@ namespace Qpid.Client.Message
 {
     public class QpidTextMessageFactory : AbstractQmsMessageFactory
     {        
-        public override AbstractQmsMessage CreateMessage()
+        public override AbstractQmsMessage CreateMessage(string mimeType)
         {
-            return new QpidTextMessage();
+            QpidTextMessage msg = new QpidTextMessage();
+            msg.ContentType = mimeType;
+            return msg;
         }
 
         protected override AbstractQmsMessage CreateMessage(long deliveryTag, ByteBuffer data, ContentHeaderBody contentHeader)
diff --git a/dotnet/Qpid.Client/Client/Message/UnprocessedMessage.cs b/dotnet/Qpid.Client/Client/Message/UnprocessedMessage.cs
index cb4e64718b..b64c8e1c27 100644
--- a/dotnet/Qpid.Client/Client/Message/UnprocessedMessage.cs
+++ b/dotnet/Qpid.Client/Client/Message/UnprocessedMessage.cs
@@ -43,7 +43,7 @@ namespace Qpid.Client.Message
             Bodies.Add(body);
             if (body.Payload != null)
             {
-                _bytesReceived += (uint)body.Payload.Length;
+                _bytesReceived += (uint)body.Payload.Remaining;
             }
         }
 
diff --git a/dotnet/Qpid.Client/Client/QpidConnectionInfo.cs b/dotnet/Qpid.Client/Client/QpidConnectionInfo.cs
index 914170467a..d88683f7d5 100644
--- a/dotnet/Qpid.Client/Client/QpidConnectionInfo.cs
+++ b/dotnet/Qpid.Client/Client/QpidConnectionInfo.cs
@@ -171,19 +171,10 @@ namespace Qpid.Client
                 sb.Append('?');
                 foreach (String key in options.Keys)
                 {
-                    sb.Append(key);
-
-                    sb.Append("='");
-
-                    sb.Append(options[key]);
-
-                    sb.Append("'");
-                    sb.Append(DEFAULT_OPTION_SEPERATOR);
+                    sb.AppendFormat("{0}='{1}'{2}", key, options[key], DEFAULT_OPTION_SEPERATOR);
                 }
 
                 sb.Remove(sb.Length - 1, 1);
-                //                sb.deleteCharAt(sb.length() - 1);
-
                 return sb.ToString();
             }
         }
@@ -358,9 +349,10 @@ namespace Qpid.Client
 
     public class QpidConnectionInfo : IConnectionInfo
     {
+        const string DEFAULT_VHOST = "/";
         string _username = "guest";
         string _password = "guest";
-        string _virtualHost = "/";
+        string _virtualHost = DEFAULT_VHOST;
 
         string _failoverMethod = null;
         IDictionary _failoverOptions = new Hashtable();
@@ -385,15 +377,51 @@ namespace Qpid.Client
 
         public string AsUrl()
         {
-            string result = "amqp://";
-            foreach (IBrokerInfo info in _brokerInfos)
+            StringBuilder sb = new StringBuilder();
+            sb.AppendFormat("{0}://", ConnectionUrlConstants.AMQ_PROTOCOL);
+
+            if (_username != null)
             {
-                result += info.ToString();
+                sb.Append(_username);
+                if (_password != null)
+                {
+                    sb.AppendFormat(":{0}", _password);
+                }
+                sb.Append("@");
             }
-            return result;
 
+            sb.Append(_clientName);
+            sb.Append(_virtualHost);
+            sb.Append(OptionsToString());
+
+            return sb.ToString();
         }
 
+        private String OptionsToString()
+        {
+            StringBuilder sb = new StringBuilder();
+            sb.AppendFormat("?{0}='", ConnectionUrlConstants.OPTIONS_BROKERLIST);
+
+            foreach (IBrokerInfo broker in _brokerInfos)
+            {
+                sb.AppendFormat("{0};", broker);
+            }
+
+            sb.Remove(sb.Length - 1, 1);
+            sb.Append("'");
+
+            if (_failoverMethod != null)
+            {
+                sb.AppendFormat("{0}{1}='{2}{3}'", URLHelper.DEFAULT_OPTION_SEPERATOR,
+                    ConnectionUrlConstants.OPTIONS_FAILOVER,
+                    _failoverMethod,
+                    URLHelper.printOptions((Hashtable)_failoverOptions));
+            }
+
+            return sb.ToString();
+        }
+
+
         public string FailoverMethod
         {
             get { return _failoverMethod; }
@@ -449,7 +477,13 @@ namespace Qpid.Client
         public string VirtualHost
         {
             get { return _virtualHost; }
-            set { _virtualHost = value; }
+            set { 
+               _virtualHost = value;
+               if ( _virtualHost == null || _virtualHost.Length == 0 )
+                  _virtualHost = DEFAULT_VHOST;
+               if ( _virtualHost[0] != '/' )
+                  _virtualHost = '/' + _virtualHost;
+            }
         }
 
         public string GetOption(string key)
diff --git a/dotnet/Qpid.Client/Client/State/AMQStateManager.cs b/dotnet/Qpid.Client/Client/State/AMQStateManager.cs
index 7be17a1080..1233f9d836 100644
--- a/dotnet/Qpid.Client/Client/State/AMQStateManager.cs
+++ b/dotnet/Qpid.Client/Client/State/AMQStateManager.cs
@@ -65,6 +65,8 @@ namespace Qpid.Client.State
             IStateAwareMethodListener channelClose = new ChannelCloseMethodHandler();
             IStateAwareMethodListener basicDeliver = new BasicDeliverMethodHandler();
             IStateAwareMethodListener basicReturn = new BasicReturnMethodHandler();
+            IStateAwareMethodListener queueDeleteOk = new QueueDeleteOkMethodHandler();
+            IStateAwareMethodListener queuePurgeOk = new QueuePurgeOkMethodHandler();
             
             // We need to register a map for the null (i.e. all state) handlers otherwise you get
             // a stack overflow in the handler searching code when you present it with a frame for which
@@ -96,6 +98,8 @@ namespace Qpid.Client.State
                 open[typeof(ConnectionCloseBody)] = connectionClose;
                 open[typeof(BasicDeliverBody)] = basicDeliver;
                 open[typeof(BasicReturnBody)] = basicReturn;
+                open[typeof(QueueDeleteOkBody)] = queueDeleteOk;
+                open[typeof(QueuePurgeOkBody)] = queuePurgeOk;
                 _state2HandlersMap[AMQState.CONNECTION_OPEN] = open;
             }
             {
diff --git a/dotnet/Qpid.Client/Client/Transport/AmqpChannel.cs b/dotnet/Qpid.Client/Client/Transport/AmqpChannel.cs
index ca7ffad8b3..4e4ca03322 100644
--- a/dotnet/Qpid.Client/Client/Transport/AmqpChannel.cs
+++ b/dotnet/Qpid.Client/Client/Transport/AmqpChannel.cs
@@ -33,35 +33,39 @@ namespace Qpid.Client.Transport
         // Warning: don't use this log for regular logging.
         static readonly ILog _protocolTraceLog = LogManager.GetLogger("Qpid.Client.ProtocolChannel.Tracing");
         
-        IByteChannel byteChannel;
-        IProtocolEncoder encoder;
-        IProtocolDecoder decoder;
+        IByteChannel _byteChannel;
+        IProtocolEncoder _encoder;
+        IProtocolDecoder _decoder;
+        IProtocolDecoderOutput _decoderOutput;
+        private object _syncLock;
 
-        public AmqpChannel(IByteChannel byteChannel)
+        public AmqpChannel(IByteChannel byteChannel, IProtocolDecoderOutput decoderOutput)
         {
-            this.byteChannel = byteChannel;    
+            _byteChannel = byteChannel;
+            _decoderOutput = decoderOutput;
+            _syncLock = new object();
             
             AMQProtocolProvider protocolProvider = new AMQProtocolProvider();
             IProtocolCodecFactory factory = protocolProvider.CodecFactory;
-            encoder = factory.Encoder;
-            decoder = factory.Decoder;
+            _encoder = factory.Encoder;
+            _decoder = factory.Decoder;
         }
 
-        public Queue Read()
+        public void Read()
         {
-            ByteBuffer buffer = byteChannel.Read();
-            return DecodeAndTrace(buffer);
+            ByteBuffer buffer = _byteChannel.Read();
+            Decode(buffer);
         }
         
         public IAsyncResult BeginRead(AsyncCallback callback, object state)
         {
-           return byteChannel.BeginRead(callback, state);
+           return _byteChannel.BeginRead(callback, state);
         }
 
-        public Queue EndRead(IAsyncResult result)
+        public void EndRead(IAsyncResult result)
         {
-           ByteBuffer buffer = byteChannel.EndRead(result);
-           return DecodeAndTrace(buffer);
+           ByteBuffer buffer = _byteChannel.EndRead(result);
+           Decode(buffer);
         }
 
         public void Write(IDataBlock o)
@@ -74,43 +78,32 @@ namespace Qpid.Client.Transport
             // we should be doing an async write, but apparently
             // the mentalis library doesn't queue async read/writes
             // correctly and throws random IOException's. Stay sync for a while
-            //byteChannel.BeginWrite(Encode(o), OnAsyncWriteDone, null);
-            byteChannel.Write(Encode(o));
+            //_byteChannel.BeginWrite(Encode(o), OnAsyncWriteDone, null);
+            _byteChannel.Write(Encode(o));
         }
 
         private void OnAsyncWriteDone(IAsyncResult result)
         {
-           byteChannel.EndWrite(result);
+           _byteChannel.EndWrite(result);
         }
 
-        private Queue DecodeAndTrace(ByteBuffer buffer)
+        private void Decode(ByteBuffer buffer)
         {
-           Queue frames = Decode(buffer);
-
-           // TODO: Refactor to decorator.
-           if ( _protocolTraceLog.IsDebugEnabled )
+           // make sure we don't try to decode more than
+           // one buffer at the same time
+           lock ( _syncLock )
            {
-              foreach ( object o in frames )
-              {
-                 _protocolTraceLog.Debug(String.Format("READ {0}", o));
-              }
+              _decoder.Decode(buffer, _decoderOutput);
            }
-           return frames;
         }
 
         private ByteBuffer Encode(object o)
         {
             SingleProtocolEncoderOutput output = new SingleProtocolEncoderOutput();
-            encoder.Encode(o, output);
+            _encoder.Encode(o, output);
             return output.buffer;
         }
 
-        private Queue Decode(ByteBuffer byteBuffer)
-        {
-            SimpleProtocolDecoderOutput outx = new SimpleProtocolDecoderOutput();
-            decoder.Decode(byteBuffer, outx);
-            return outx.MessageQueue;
-        }
     }
 }
 
diff --git a/dotnet/Qpid.Client/Client/Transport/IProtocolChannel.cs b/dotnet/Qpid.Client/Client/Transport/IProtocolChannel.cs
index 0379e582d6..e4d4d2ed29 100644
--- a/dotnet/Qpid.Client/Client/Transport/IProtocolChannel.cs
+++ b/dotnet/Qpid.Client/Client/Transport/IProtocolChannel.cs
@@ -25,8 +25,8 @@ namespace Qpid.Client.Transport
 {
     public interface IProtocolChannel : IProtocolWriter
     {
-        Queue Read();
+        void Read();
         IAsyncResult BeginRead(AsyncCallback callback, object state);
-        Queue EndRead(IAsyncResult result);
+        void EndRead(IAsyncResult result);
     }
 }
diff --git a/dotnet/Qpid.Client/Client/Transport/ProtocolDecoderOutput.cs b/dotnet/Qpid.Client/Client/Transport/ProtocolDecoderOutput.cs
new file mode 100644
index 0000000000..07df62ea84
--- /dev/null
+++ b/dotnet/Qpid.Client/Client/Transport/ProtocolDecoderOutput.cs
@@ -0,0 +1,59 @@
+/*

+ *

+ * 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.Threading;

+using Qpid.Client.Protocol;

+using Qpid.Codec;

+using Qpid.Framing;

+using log4net;

+

+namespace Qpid.Client.Transport

+{

+   /// <summary>

+   /// <see cref="IProtocolDecoderOutput"/> implementation that forwards

+   /// each <see cref="IDataBlock"/> as it is decoded to the

+   /// protocol listener

+   /// </summary>

+   internal class ProtocolDecoderOutput : IProtocolDecoderOutput

+   {

+      private IProtocolListener _protocolListener;

+      static readonly ILog _protocolTraceLog = LogManager.GetLogger("Qpid.Client.ProtocolChannel.Tracing");

+

+      public ProtocolDecoderOutput(IProtocolListener protocolListener)

+      {

+         if ( protocolListener == null )

+            throw new ArgumentNullException("protocolListener");

+

+         _protocolListener = protocolListener;

+      }

+

+      public void Write(object message)

+      {

+         IDataBlock block = message as IDataBlock;

+         if ( block != null )

+         {

+            _protocolTraceLog.Debug(String.Format("READ {0}", block));

+            _protocolListener.OnMessage(block);

+         }

+      }

+   }

+} // namespace Qpid.Client.Transport

+

diff --git a/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs b/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs
index 1fb07fb245..2895c75431 100644
--- a/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs
+++ b/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs
@@ -24,6 +24,7 @@ using System.IO;
 using System.Threading;

 using Qpid.Client.Qms;

 using Qpid.Client.Protocol;

+using Qpid.Codec;

 using Qpid.Framing;

 

 namespace Qpid.Client.Transport.Socket.Blocking

@@ -66,7 +67,11 @@ namespace Qpid.Client.Transport.Socket.Blocking
          _ioHandler = MakeBrokerConnection(broker, connection);

          // todo: get default read size from config!

 

-         _amqpChannel = new AmqpChannel(new ByteChannel(_ioHandler));

+         IProtocolDecoderOutput decoderOutput =

+            new ProtocolDecoderOutput(_protocolListener);

+         _amqpChannel = 

+            new AmqpChannel(new ByteChannel(_ioHandler), decoderOutput);

+

          // post an initial async read

          _amqpChannel.BeginRead(new AsyncCallback(OnAsyncReadDone), this);

       }

@@ -117,22 +122,28 @@ namespace Qpid.Client.Transport.Socket.Blocking
       {

          try

          {

-            Queue frames = _amqpChannel.EndRead(result);

-

-            // process results

-            foreach ( IDataBlock dataBlock in frames )

-            {

-               _protocolListener.OnMessage(dataBlock);

-            }

-            // if we're not stopping, post a read again

+            _amqpChannel.EndRead(result);

+

             bool stopping = _stopEvent.WaitOne(0, false);

             if ( !stopping )

                _amqpChannel.BeginRead(new AsyncCallback(OnAsyncReadDone), null);

          } catch ( Exception e )

          {

-            _protocolListener.OnException(e);

+            // ignore any errors during closing

+            bool stopping = _stopEvent.WaitOne(0, false);

+            if ( !stopping )

+               _protocolListener.OnException(e);

          }

       }

+

+      #region IProtocolDecoderOutput Members

+

+      public void Write(object message)

+      {

+         _protocolListener.OnMessage((IDataBlock)message);

+      }

+

+      #endregion

    }

 }