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

1 files changed, 113 insertions, 100 deletions

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;
+      }
+   }
 }