* 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)


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

23 files changed, 684 insertions, 705 deletions

diff --git a/dotnet/Qpid.Buffer.Tests/SimpleByteBufferTests.cs b/dotnet/Qpid.Buffer.Tests/SimpleByteBufferTests.cs
index 290e908a0d..aa675e3014 100644
--- a/dotnet/Qpid.Buffer.Tests/SimpleByteBufferTests.cs
+++ b/dotnet/Qpid.Buffer.Tests/SimpleByteBufferTests.cs
@@ -120,6 +120,9 @@ namespace Qpid.Buffer.Tests
          Assert.AreEqual(10, buffer.Limit);

          Assert.AreEqual(2, buffer.Position);

          Assert.AreEqual(8, buffer.Remaining);

+         buffer.Rewind();

+         Assert.AreEqual((byte)0x02, buffer.GetByte());

+         Assert.AreEqual((byte)0x03, buffer.GetByte());

       }

 

       [Test]

@@ -326,3 +329,4 @@ namespace Qpid.Buffer.Tests
 

    } // class SimpleByteBufferTests

 }

+

diff --git a/dotnet/Qpid.Buffer/ByteBuffer.cs b/dotnet/Qpid.Buffer/ByteBuffer.cs
index 1bfc749746..b341fa7632 100644
--- a/dotnet/Qpid.Buffer/ByteBuffer.cs
+++ b/dotnet/Qpid.Buffer/ByteBuffer.cs
@@ -809,7 +809,7 @@ namespace Qpid.Buffer
       /// </summary>
       /// <param name="position">Position to read from</param>
       /// <returns>The value at the position</returns>
-      public char getChar(int position)
+      public char GetChar(int position)
       {
          return (char)GetUInt16(position);
       }
@@ -941,7 +941,7 @@ namespace Qpid.Buffer
       {
          if ( position + length > Limit )
          {
-            throw new BufferUnderflowException("Attempt to read " + length + " byte(s) to buffer where position is " + _position +
+            throw new BufferUnderflowException("Attempt to read " + length + " byte(s) to buffer where position is " + position +
                                                " and limit is " + Limit);
          }
       }
@@ -954,7 +954,7 @@ namespace Qpid.Buffer
          }
          if ( position + length > Limit )
          {
-            throw new BufferOverflowException("Attempt to write " + length + " byte(s) to buffer where position is " + _position +
+            throw new BufferOverflowException("Attempt to write " + length + " byte(s) to buffer where position is " + position +
                                               " and limit is " + Limit);
          }
       }
diff --git a/dotnet/Qpid.Client.Tests/MultiConsumer/ProducerMultiConsumer.cs b/dotnet/Qpid.Client.Tests/MultiConsumer/ProducerMultiConsumer.cs
index 687f08eeef..fd1400d9d8 100644
--- a/dotnet/Qpid.Client.Tests/MultiConsumer/ProducerMultiConsumer.cs
+++ b/dotnet/Qpid.Client.Tests/MultiConsumer/ProducerMultiConsumer.cs
@@ -38,7 +38,7 @@ namespace Qpid.Client.Tests
 
         private const int MESSAGE_COUNT = 1000;
 
-        private const string MESSAGE_DATA_BYTES = "jfd ghljgl hjvhlj cvhvjf ldhfsj lhfdsjf hldsjfk hdslkfj hsdflk  ";
+        private const string MESSAGE_DATA_BYTES = "****jfd ghljgl hjvhlj cvhvjf ldhfsj lhfdsjf hldsjfk hdslkfj hsdflk  ";
 
         AutoResetEvent _finishedEvent = new AutoResetEvent(false);
 
@@ -100,7 +100,9 @@ namespace Qpid.Client.Tests
                 _logger.Info("All messages received");
                 _finishedEvent.Set();
             }
-        }
+            if ( newCount % 100 == 0 ) 
+               System.Diagnostics.Debug.WriteLine(((ITextMessage)m).Text);
+         }
         
         [Test]
         public void RunTest()
@@ -110,7 +112,7 @@ namespace Qpid.Client.Tests
                 ITextMessage msg;
                 try
                 {
-                    msg = _channel.CreateTextMessage(GetData(512 + 8*i));
+                   msg = _channel.CreateTextMessage(GetData(512 + 8*i));
                 }
                 catch (Exception e)
                 {
diff --git a/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj b/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj
index 1a930b178b..4f73335e7e 100644
--- a/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj
+++ b/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj
@@ -44,7 +44,6 @@
     <Reference Include="System.Xml" />

   </ItemGroup>

   <ItemGroup>

-    <Compile Include="bio\BlockingIo.cs" />

     <Compile Include="BrokerDetails\BrokerDetailsTest.cs" />

     <Compile Include="Channel\ChannelMessageCreationTests.cs" />

     <Compile Include="Messages\MessageFactoryRegistryTests.cs" />

diff --git a/dotnet/Qpid.Client.Tests/bio/BlockingIo.cs b/dotnet/Qpid.Client.Tests/bio/BlockingIo.cs
deleted file mode 100644
index 24f3299dae..0000000000
--- a/dotnet/Qpid.Client.Tests/bio/BlockingIo.cs
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- *
- * 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 NUnit.Framework;
-using Qpid.Client.Protocol;
-using Qpid.Framing;
-using Qpid.Messaging;
-
-namespace Qpid.Client.Transport
-{
-    [TestFixture]
-    public class BlockingIo
-    {
-        [Test]
-        public void connectFromOutside()
-        {
-            QpidConnectionInfo connectionInfo = new QpidConnectionInfo();
-            connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 5672, false));
-            AMQConnection connection = new AMQConnection(connectionInfo);
-            ProtocolWriter protocolWriter = connection.ConvenientProtocolWriter;
-
-            // TODO: Open channels and handle them simultaneously.
-            // Need more thread here?
-            // Send ChannelOpen.
-            ushort channelId = 1;
-            protocolWriter.SyncWrite(ChannelOpenBody.CreateAMQFrame(channelId, null), typeof (ChannelOpenOkBody));
-
-            connection.Close();
-        }
-
-        [Test]
-        public void regularConnection()
-        {
-            QpidConnectionInfo connectionInfo = new QpidConnectionInfo();
-            connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 5672, false));
-            using (IConnection connection = new AMQConnection(connectionInfo)) {
-                Console.WriteLine("connection = {0}", connection);
-                Thread.Sleep(2000);
-            }
-        }
-
-        [Test]
-        public void connectionAndSleepForHeartbeats()
-        {
-            QpidConnectionInfo connectionInfo = new QpidConnectionInfo();
-            connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost", 5672, false));
-            using (IConnection connection = new AMQConnection(connectionInfo))
-            {
-                Console.WriteLine("connection = {0}", connection);
-                Thread.Sleep(60000);
-            }
-        }
-    }
-}
diff --git a/dotnet/Qpid.Client.Tests/log4net.config b/dotnet/Qpid.Client.Tests/log4net.config
index e5340a1500..4346e0eaeb 100644
--- a/dotnet/Qpid.Client.Tests/log4net.config
+++ b/dotnet/Qpid.Client.Tests/log4net.config
@@ -49,8 +49,12 @@
     <level value="info" />
     <appender-ref ref="ioLog"/>
   </logger>
+   <logger name="Qpid.Framing.FieldTable" additivity="false">
+      <level value="debug" />
+      <appender-ref ref="console"/>
+   </logger>
 
-  <root>
+   <root>
     <appender-ref ref="console"/>
     <appender-ref ref="UdpAppender"/>
 	  <appender-ref ref="filelog"/>
diff --git a/dotnet/Qpid.Client.Tests/requestreply1/ServiceProvidingClient.cs b/dotnet/Qpid.Client.Tests/requestreply1/ServiceProvidingClient.cs
index 515ae41e1c..ad5981a5c5 100644
--- a/dotnet/Qpid.Client.Tests/requestreply1/ServiceProvidingClient.cs
+++ b/dotnet/Qpid.Client.Tests/requestreply1/ServiceProvidingClient.cs
@@ -35,16 +35,15 @@ namespace Qpid.Client.Tests
 
         private string _replyToExchangeName;
         private string _replyToRoutingKey;
+        const int PACK = 100;
 
         private IMessagePublisher _destinationPublisher;
+        private IMessageConsumer _consumer;
 
         private string _serviceName = "ServiceQ1";
 
         private string _selector = null;
 
-        //private EventWaitHandle _event = new ManualResetEvent(false);
-        private AutoResetEvent _event = new AutoResetEvent(false);
-
         [SetUp]
         public override void Init()
         {            
@@ -59,36 +58,38 @@ namespace Qpid.Client.Tests
             
             _channel.DeclareQueue(_serviceName, false, false, false);
 
-            IMessageConsumer consumer = _channel.CreateConsumerBuilder(_serviceName)
+            _consumer = _channel.CreateConsumerBuilder(_serviceName)
                 .WithPrefetchLow(100)
                 .WithPrefetchHigh(500)
                 .WithNoLocal(true)
                 .Create();
-            consumer.OnMessage = new MessageReceivedDelegate(OnMessage);
+            _consumer.OnMessage = new MessageReceivedDelegate(OnMessage);
+        }
+
+        public override void Shutdown()
+        {
+           _consumer.Dispose();
+           base.Shutdown();
         }
 
         private void OnConnectionException(Exception e)
         {
             _logger.Info("Connection exception occurred", e);
-            _event.Set(); // Shutdown test on error
             // XXX: Test still doesn't shutdown when broker terminates. Is there no heartbeat?
         }
 
         [Test]
-        public void TestFail()
-        {
-            Assert.Fail("Tests in this class do not run on autopilot, but hang forever, so commented out until can be fixed.");
-        }
-
-        /*[Test]
         public void Test()
         {
             _connection.Start();
             _logger.Info("Waiting...");
-            _event.WaitOne();
-        }*/
 
-        public void OnMessage(IMessage message)
+            ServiceRequestingClient client = new ServiceRequestingClient();
+            client.Init();
+            client.SendMessages();
+        }
+
+        private void OnMessage(IMessage message)
         {
 //            _logger.Info("Got message '" + message + "'");
 
@@ -109,9 +110,9 @@ namespace Qpid.Client.Tests
                 _destinationPublisher = _channel.CreatePublisherBuilder()
                     .WithExchangeName(_replyToExchangeName)
                     .WithRoutingKey(_replyToRoutingKey)
+                    .WithDeliveryMode(DeliveryMode.NonPersistent)
                     .Create();
                 _destinationPublisher.DisableMessageTimestamp = true;
-                _destinationPublisher.DeliveryMode = DeliveryMode.NonPersistent;
                 _logger.Debug("After create a producer");
             }
             catch (QpidException e)
@@ -120,7 +121,7 @@ namespace Qpid.Client.Tests
                 throw e;
             }
             _messageCount++;
-            if (_messageCount % 1000 == 0)
+            if (_messageCount % PACK == 0)
             {
                 _logger.Info("Received message total: " + _messageCount);
                 _logger.Info(string.Format("Sending response to '{0}:{1}'", 
@@ -129,25 +130,20 @@ namespace Qpid.Client.Tests
 
             try
             {
-                String payload = "This is a response: sing together: 'Mahnah mahnah...'" + tm.Text;
-                ITextMessage msg = _channel.CreateTextMessage(payload);
-                if (tm.Headers.Contains("timeSent"))
-                {
-//                    _logger.Info("timeSent property set on message");
-//                    _logger.Info("timeSent value is: " + tm.Headers["timeSent"]);
-                    msg.Headers["timeSent"] = tm.Headers["timeSent"];
-                }
-                _destinationPublisher.Send(msg);
-                if (_messageCount % 1000 == 0)
-                {
-                    _logger.Info(string.Format("Sending response to '{0}:{1}'",
-                                               _replyToExchangeName, _replyToRoutingKey));
-                }
-            }
-            catch (QpidException e)
+               String payload = "This is a response: sing together: 'Mahnah mahnah...'" + tm.Text;
+               ITextMessage msg = _channel.CreateTextMessage(payload);
+               if ( tm.Headers.Contains("timeSent") )
+               {
+                  msg.Headers["timeSent"] = tm.Headers["timeSent"];
+               }
+               _destinationPublisher.Send(msg);
+            } catch ( QpidException e )
             {
-                _logger.Error("Error sending message: " + e, e);
-                throw e;
+               _logger.Error("Error sending message: " + e, e);
+               throw e;
+            } finally
+            {
+               _destinationPublisher.Dispose();
             }
         }               
     }
diff --git a/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs b/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs
index 4479d767ea..8264879c1f 100644
--- a/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs
+++ b/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs
@@ -26,18 +26,18 @@ using Qpid.Messaging;
 
 namespace Qpid.Client.Tests
 {
-    [TestFixture]
     public class ServiceRequestingClient : BaseMessagingTestFixture
     {
         private const int MESSAGE_SIZE = 1024;
         private static string MESSAGE_DATA = new string('x', MESSAGE_SIZE);
 
-        private const int NUM_MESSAGES = 10000;
+        private const int PACK = 100;
+        private const int NUM_MESSAGES = PACK*10; // increase when in standalone
 
         private static ILog _log = LogManager.GetLogger(typeof(ServiceRequestingClient));
 
-        AutoResetEvent _finishedEvent = new AutoResetEvent(false);
-
+        ManualResetEvent _finishedEvent = new ManualResetEvent(false);
+        
         private int _expectedMessageCount = NUM_MESSAGES;
 
         private long _startTime = 0;        
@@ -54,9 +54,9 @@ namespace Qpid.Client.Tests
             {
                 _publisher = _channel.CreatePublisherBuilder()
                     .WithRoutingKey(_commandQueueName)
+                    .WithDeliveryMode(DeliveryMode.NonPersistent)
                     .Create();
                 _publisher.DisableMessageTimestamp = true; // XXX: need a "with" for this in builder?
-                _publisher.DeliveryMode = DeliveryMode.NonPersistent;  // XXX: need a "with" for this in builder?
             }
             catch (QpidException e)
             {
@@ -64,7 +64,7 @@ namespace Qpid.Client.Tests
             }
         }
         
-        /*[Test]
+        [Test]
         public void SendMessages()
         {
             InitialiseProducer();
@@ -100,47 +100,18 @@ namespace Qpid.Client.Tests
                 // Added timestamp.
                 long timeNow = DateTime.Now.Ticks;
                 string timeSentString = String.Format("{0:G}", timeNow);
-//                _log.Info(String.Format("timeSent={0} timeSentString={1}", timeNow, timeSentString));
-                msg.Headers.SetString("timeSent", timeSentString);
-                //msg.Headers.SetLong("sentAt", timeNow);
+                msg.Headers.SetLong("timeSent", timeNow);
                 
-                try
-                {
-                    _publisher.Send(msg);
-                }
-                catch (Exception e)
-                {
-                    _log.Error("Error sending message: " + e, e);
-                    //base._port = 5673;
-                    _log.Info("Reconnecting but on port 5673");
-                    try
-                    {
-                        base.Init();
-                        InitialiseProducer();
-                        // cheesy but a quick test
-                        _log.Info("Calling SendMessages again");
-                        SendMessages();
-                    }
-                    catch (Exception ex)
-                    {
-                        _log.Error("Totally busted: failed to reconnect: " + ex, ex);
-                    }
-                }
+                 _publisher.Send(msg);
             }
 
             // Assert that the test finishes within a reasonable amount of time.
-            const int waitSeconds = 10;
+            const int waitSeconds = 40;
             const int waitMilliseconds = waitSeconds * 1000;
             _log.Info("Finished sending " + _expectedMessageCount + " messages");
             _log.Info(String.Format("Waiting {0} seconds to receive last message...", waitSeconds));
             Assert.IsTrue(_finishedEvent.WaitOne(waitMilliseconds, false), 
                           String.Format("Expected to finish in {0} seconds", waitSeconds));
-        }*/
-
-        [Test]
-        public void TestFail()
-        {
-            Assert.Fail("Tests in this class do not run on autopilot, but hang forever, so commented out until can be fixed.");
         }
 
         public void OnMessage(IMessage m)
@@ -150,23 +121,19 @@ namespace Qpid.Client.Tests
                 _log.Debug("Message received: " + m);
             }
 
-            //if (m.Headers.Contains("sentAt"))
             if (!m.Headers.Contains("timeSent"))
             {
                 throw new Exception("Set timeSent!");
             }
-            //long sentAt = m.Headers.GetLong("sentAt");
-            long sentAt = Int64.Parse(m.Headers.GetString("timeSent"));
+
+            long sentAt = m.Headers.GetLong("timeSent");
             long now = DateTime.Now.Ticks;
             long latencyTicks = now - sentAt;
-//                _log.Info(String.Format("latency = {0} ticks ", latencyTicks));
             long latencyMilliseconds = latencyTicks / TimeSpan.TicksPerMillisecond;
-//                _log.Info(String.Format("latency = {0} ms", latencyMilliseconds));
 
             averager.Add(latencyMilliseconds);
 
-            // Output average every 1000 messages.
-            if (averager.Num % 1000 == 0)
+            if (averager.Num % PACK == 0)
             {
                 _log.Info("Ticks per millisecond = " + TimeSpan.TicksPerMillisecond);
                 _log.Info(String.Format("Average latency (ms) = {0}", averager));
@@ -185,13 +152,6 @@ namespace Qpid.Client.Tests
                 _finishedEvent.Set(); // Notify main thread to quit.
             }
         }
-        
-        /*public static void Main(String[] args)
-        {
-            ServiceRequestingClient c = new ServiceRequestingClient();
-            c.Init();
-            c.SendMessages();
-        }*/
     }
     
     class Avergager
diff --git a/dotnet/Qpid.Client/Client/AmqChannel.cs b/dotnet/Qpid.Client/Client/AmqChannel.cs
index 0d0c48303a..56202e0094 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;
 
@@ -763,119 +763,6 @@ namespace Qpid.Client
             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)
-            {
-                currentTime = DateTime.UtcNow.Ticks;
-                message.Timestamp = currentTime;
-            }
-
-            ByteBuffer buf = message.Data;
-            byte[] payload = null;
-            if (buf != null)
-            {
-                payload = new byte[buf.Remaining];
-                buf.GetBytes(payload);
-            }
-            BasicContentHeaderProperties contentHeaderProperties = message.ContentHeaderProperties;
-
-            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));
-            }
-
-            // 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)
-            {
-                _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 (_connection.FailoverMutex) {
-                _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(byte[] payload)
-        {
-            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;
-        }
-
         public string GenerateUniqueName()
         {
             string result = _connection.ProtocolSession.GenerateQueueName();            
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/Message/AMQMessageFactory.cs b/dotnet/Qpid.Client/Client/Message/AMQMessageFactory.cs
index 61b308696c..a7ee085a04 100644
--- a/dotnet/Qpid.Client/Client/Message/AMQMessageFactory.cs
+++ b/dotnet/Qpid.Client/Client/Message/AMQMessageFactory.cs
@@ -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/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/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/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

    }

 }

 

diff --git a/dotnet/Qpid.Client/Qpid.Client.csproj b/dotnet/Qpid.Client/Qpid.Client.csproj
index 2bafeb23f6..de585be7e0 100644
--- a/dotnet/Qpid.Client/Qpid.Client.csproj
+++ b/dotnet/Qpid.Client/Qpid.Client.csproj
@@ -110,6 +110,7 @@
     <Compile Include="Client\Transport\IProtocolChannel.cs" />

     <Compile Include="Client\Transport\IProtocolWriter.cs" />

     <Compile Include="Client\Transport\ITransport.cs" />

+    <Compile Include="Client\Transport\ProtocolDecoderOutput.cs" />

     <Compile Include="Client\Transport\SingleProtocolEncoderOutput.cs" />

     <Compile Include="Client\Transport\Socket\Blocking\BlockingSocketTransport.cs" />

     <Compile Include="Client\Transport\Socket\Blocking\ByteChannel.cs" />

diff --git a/dotnet/Qpid.Codec/CumulativeProtocolDecoder.cs b/dotnet/Qpid.Codec/CumulativeProtocolDecoder.cs
index 72c56e0b17..3d454b284b 100644
--- a/dotnet/Qpid.Codec/CumulativeProtocolDecoder.cs
+++ b/dotnet/Qpid.Codec/CumulativeProtocolDecoder.cs
@@ -19,121 +19,134 @@
  *
  */
 using System;
+using log4net;
 using Qpid.Buffer;
 
 namespace Qpid.Codec
 {
-    public abstract class CumulativeProtocolDecoder : IProtocolDecoder
-    {
-        ByteBuffer _remaining;
+   public abstract class CumulativeProtocolDecoder : IProtocolDecoder
+   {
+      static ILog _logger = LogManager.GetLogger(typeof(CumulativeProtocolDecoder));
 
-        /// <summary>
-        /// Creates a new instance with the 4096 bytes initial capacity of
-        /// cumulative buffer.
-        /// </summary>
-        protected CumulativeProtocolDecoder()
-        {
-            _remaining = ByteBuffer.Allocate(4096);
-            _remaining.IsAutoExpand = true;
-        }
+      ByteBuffer _remaining;
 
-        /// <summary>
-        /// Cumulates content of <tt>in</tt> into internal buffer and forwards
-        /// decoding request to {@link #doDecode(IoSession, ByteBuffer, ProtocolDecoderOutput)}.
-        /// <tt>doDecode()</tt> is invoked repeatedly until it returns <tt>false</tt>
-        /// and the cumulative buffer is compacted after decoding ends.
-        /// </summary>
-        /// <exception cref="Exception">
-        /// if your <tt>doDecode()</tt> returned <tt>true</tt> not consuming the cumulative buffer.
-        /// </exception>
-        public void Decode(ByteBuffer input, IProtocolDecoderOutput output)
-        {
-            if (_remaining.Position != 0) // If there were remaining undecoded bytes
-            {
-                DecodeRemainingAndInput(input, output);
-            }
-            else
-            {
-                DecodeInput(input, output);
-            }
-        }
+      /// <summary>
+      /// Creates a new instance with the 4096 bytes initial capacity of
+      /// cumulative buffer.
+      /// </summary>
+      protected CumulativeProtocolDecoder()
+      {
+         _remaining = AllocateBuffer();
+      }
 
-        private void DecodeInput(ByteBuffer input, IProtocolDecoderOutput output)
-        {
-            // Just decode the input buffer and remember any remaining undecoded bytes.
-            try
-            {
-                DecodeAll(input, output);
-            }
-            finally
+      /// <summary>
+      /// Cumulates content of <tt>in</tt> into internal buffer and forwards
+      /// decoding request to {@link #doDecode(IoSession, ByteBuffer, ProtocolDecoderOutput)}.
+      /// <tt>doDecode()</tt> is invoked repeatedly until it returns <tt>false</tt>
+      /// and the cumulative buffer is compacted after decoding ends.
+      /// </summary>
+      /// <exception cref="Exception">
+      /// if your <tt>doDecode()</tt> returned <tt>true</tt> not consuming the cumulative buffer.
+      /// </exception>
+      public void Decode(ByteBuffer input, IProtocolDecoderOutput output)
+      {
+         if ( _remaining.Position != 0 ) // If there were remaining undecoded bytes
+         {
+            DecodeRemainingAndInput(input, output);
+         } else
+         {
+            DecodeInput(input, output);
+         }
+      }
+
+      private void DecodeInput(ByteBuffer input, IProtocolDecoderOutput output)
+      {
+         _logger.Debug(string.Format("DecodeInput: input {0}", input.Remaining));
+         // Just decode the input buffer and remember any remaining undecoded bytes.
+         try
+         {
+            DecodeAll(input, output);
+         } finally
+         {
+            if ( input.HasRemaining )
             {
-                if (input.HasRemaining)
-                {
-                    _remaining.Put(input);
-                }
+               _remaining.Put(input);
             }
-        }
+         }
+      }
 
-        private void DecodeRemainingAndInput(ByteBuffer input, IProtocolDecoderOutput output)
-        {
-            // Concatenate input buffer with left-over bytes.
-            _remaining.Put(input);
-            _remaining.Flip();
+      private void DecodeRemainingAndInput(ByteBuffer input, IProtocolDecoderOutput output)
+      {
+         _logger.Debug(string.Format("DecodeRemainingAndInput: input {0}, remaining {1}", input.Remaining, _remaining.Position));
+         // replace the _remainder buffer, so that we can leave the 
+         // original one alone. Necessary because some consumer splice
+         // the buffer and only consume it until later, causing
+         // a race condition if we compact it too soon.
+         ByteBuffer newRemainding = AllocateBuffer();
+         ByteBuffer temp = _remaining;
+         _remaining = newRemainding;
+         temp.Put(input);
+         temp.Flip();
+         try
+         {
+            DecodeAll(temp, output);
+         } finally
+         {
+            if ( temp.Remaining > 0 )
+               _remaining.Put(temp);
+         }
+      }
 
-            try
-            {
-                DecodeAll(_remaining, output);
-            }
-            finally
+      private void DecodeAll(ByteBuffer buf, IProtocolDecoderOutput output)
+      {
+         for ( ; ; )
+         {
+            int oldPos = buf.Position;
+            bool decoded = DoDecode(buf, output);
+            if ( decoded )
             {
-                _remaining.Compact();
-            }
-        }
+               if ( buf.Position == oldPos )
+               {
+                  throw new Exception(
+                      "doDecode() can't return true when buffer is not consumed.");
+               }
 
-        private void DecodeAll(ByteBuffer buf, IProtocolDecoderOutput output)
-        {
-            for (;;)
+               if ( !buf.HasRemaining )
+               {
+                  break;
+               }
+            } else
             {
-                int oldPos = buf.Position;
-                bool decoded = DoDecode(buf, output);
-                if (decoded)
-                {                        
-                    if (buf.Position == oldPos)
-                    {
-                        throw new Exception(
-                            "doDecode() can't return true when buffer is not consumed.");
-                    }
-
-                    if (!buf.HasRemaining)
-                    {
-                        break;
-                    }
-                }
-                else
-                {
-                    break;
-                }
+               break;
             }
-        }
+         }
+      }
+
+      /// <summary>
+      /// Implement this method to consume the specified cumulative buffer and
+      /// decode its content into message(s). 
+      /// </summary>
+      /// <param name="input">the cumulative buffer</param>
+      /// <param name="output">decoder output</param>
+      /// <returns>
+      /// <tt>true</tt> if and only if there's more to decode in the buffer
+      /// and you want to have <tt>doDecode</tt> method invoked again.
+      /// Return <tt>false</tt> if remaining data is not enough to decode,
+      /// then this method will be invoked again when more data is cumulated.
+      /// </returns>
+      /// <exception cref="Exception">If cannot decode</exception>
+      protected abstract bool DoDecode(ByteBuffer input, IProtocolDecoderOutput output);
 
-        /// <summary>
-        /// Implement this method to consume the specified cumulative buffer and
-        /// decode its content into message(s). 
-        /// </summary>
-        /// <param name="input">the cumulative buffer</param>
-        /// <param name="output">decoder output</param>
-        /// <returns>
-        /// <tt>true</tt> if and only if there's more to decode in the buffer
-        /// and you want to have <tt>doDecode</tt> method invoked again.
-        /// Return <tt>false</tt> if remaining data is not enough to decode,
-        /// then this method will be invoked again when more data is cumulated.
-        /// </returns>
-        /// <exception cref="Exception">If cannot decode</exception>
-        protected abstract bool DoDecode(ByteBuffer input, IProtocolDecoderOutput output);
+      public void Dispose()
+      {
+         _remaining = null;
+      }
 
-        public void Dispose()
-        {
-            _remaining = null;
-        }
-    }
+      private ByteBuffer AllocateBuffer()
+      {
+         ByteBuffer buffer = ByteBuffer.Allocate(4096);
+         buffer.IsAutoExpand = true;
+         return buffer;
+      }
+   }
 }
diff --git a/dotnet/Qpid.Common/Framing/ContentBody.cs b/dotnet/Qpid.Common/Framing/ContentBody.cs
index b63df22339..617086f2e8 100644
--- a/dotnet/Qpid.Common/Framing/ContentBody.cs
+++ b/dotnet/Qpid.Common/Framing/ContentBody.cs
@@ -26,11 +26,24 @@ namespace Qpid.Framing
     {
         public const byte TYPE = 3;
 
-        /// <summary>
-        /// 
-        /// </summary>
-        /// TODO: consider whether this should be a pointer into the ByteBuffer to avoid copying */
-        public byte[] Payload;
+        private ByteBuffer _payload;
+       
+        public ByteBuffer Payload
+        {
+           get { return _payload; }
+        }
+
+       public ContentBody()
+       {
+       }
+       public ContentBody(ByteBuffer payload)
+       {
+          PopulateFromBuffer(payload, (uint)payload.Remaining);
+       }
+       public ContentBody(ByteBuffer payload, uint length)
+       {
+          PopulateFromBuffer(payload, length);
+       }
 
         #region IBody Members
 
@@ -46,7 +59,7 @@ namespace Qpid.Framing
         {
             get
             {
-                return (ushort)(Payload == null ? 0 : Payload.Length);
+                return (ushort)(Payload == null ? 0 : Payload.Remaining);
             }
         }
 
@@ -55,6 +68,7 @@ namespace Qpid.Framing
             if (Payload != null)
             {
                 buffer.Put(Payload);
+                Payload.Rewind();
             }
         }
 
@@ -62,8 +76,9 @@ namespace Qpid.Framing
         {
             if (size > 0)
             {
-                Payload = new byte[size];
-                buffer.GetBytes(Payload);
+                _payload = buffer.Slice();
+                _payload.Limit = (int)size;
+                buffer.Skip((int)size);
             }
         }
 
@@ -76,5 +91,10 @@ namespace Qpid.Framing
             frame.BodyFrame = body;
             return frame;
         }
+
+        public override string ToString()
+        {
+            return string.Format("ContentBody [ Size: {0} ]", Size);
+        }
     }
 }
diff --git a/dotnet/Qpid.Common/Framing/FieldTable.cs b/dotnet/Qpid.Common/Framing/FieldTable.cs
index fe83fff721..adff817923 100644
--- a/dotnet/Qpid.Common/Framing/FieldTable.cs
+++ b/dotnet/Qpid.Common/Framing/FieldTable.cs
@@ -47,7 +47,6 @@ namespace Qpid.Framing
         /// </summary>
         /// <param name="buffer">the buffer from which to read data. The length byte must be read already</param>
         /// <param name="length">the length of the field table. Must be > 0.</param>
-        /// <exception cref="AMQFrameDecodingException">if there is an error decoding the table</exception>
         public FieldTable(ByteBuffer buffer, uint length) : this()
         {
            _encodedForm = buffer.Slice();
@@ -497,27 +496,18 @@ namespace Qpid.Framing
         
         private AMQTypedValue GetProperty(string name)
         {
-           lock ( _syncLock )
-           {
-              if ( _properties == null )
-              {
-                 if ( _encodedForm == null )
-                 {
-                    return null;
-                 } else
-                 {
-                    PopulateFromBuffer();
-                 }
-              }
-              return (AMQTypedValue) _properties[name];
-           }
+           InitMapIfNecessary();
+           return (AMQTypedValue) _properties[name];
         }
 
         private void PopulateFromBuffer()
         {
            try
            {
-              SetFromBuffer(_encodedForm, _encodedSize);
+              ByteBuffer buffer = _encodedForm;
+              _encodedForm = null;
+              if ( buffer != null )
+                 SetFromBuffer(buffer, _encodedSize);
            } catch ( AMQFrameDecodingException e )
            {
               _log.Error("Error decoding FieldTable in deferred decoding mode ", e);
@@ -598,7 +588,11 @@ namespace Qpid.Framing
         {
            if ( _encodedForm != null )
            {
-              buffer.Put(_encodedForm);
+              lock ( _syncLock )
+              {
+                 buffer.Put(_encodedForm);
+                 _encodedForm.Flip();
+              }
            } else if ( _properties != null )
            {
               foreach ( DictionaryEntry de in _properties )
@@ -629,6 +623,7 @@ namespace Qpid.Framing
                        _log.Debug("Buffer Position:" + buffer.Position +
                                   " Remaining:" + buffer.Remaining);
                     }
+                    throw;
                  }
               }
            }
diff --git a/dotnet/Qpid.Messaging/IHeaders.cs b/dotnet/Qpid.Messaging/IHeaders.cs
index aa2d0278f7..edd0ef989b 100644
--- a/dotnet/Qpid.Messaging/IHeaders.cs
+++ b/dotnet/Qpid.Messaging/IHeaders.cs
@@ -35,7 +35,7 @@ namespace Qpid.Messaging
     {
         bool Contains(string name);
 
-        string this[string name] { get; set; }
+        object this[string name] { get; set; }
         
         bool GetBoolean(string name);
         void SetBoolean(string name, bool value);