1 files changed, 152 insertions, 0 deletions
diff --git a/qpid/dotnet/Qpid.Codec/CumulativeProtocolDecoder.cs b/qpid/dotnet/Qpid.Codec/CumulativeProtocolDecoder.cs
new file mode 100644
index 0000000000..6cfd75c851
--- /dev/null
+++ b/qpid/dotnet/Qpid.Codec/CumulativeProtocolDecoder.cs
@@ -0,0 +1,152 @@
+/*
+ *
+ * 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 log4net;
+using Apache.Qpid.Buffer;
+
+namespace Apache.Qpid.Codec
+{
+   public abstract class CumulativeProtocolDecoder : IProtocolDecoder
+   {
+      static ILog _logger = LogManager.GetLogger(typeof(CumulativeProtocolDecoder));
+
+      ByteBuffer _remaining;
+
+      /// <summary>
+      /// Creates a new instance with the 4096 bytes initial capacity of
+      /// cumulative buffer.
+      /// </summary>
+      protected CumulativeProtocolDecoder()
+      {
+         _remaining = AllocateBuffer();
+      }
+
+      /// <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 )
+            {
+               _remaining.Put(input);
+            }
+         }
+      }
+
+      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);
+         }
+      }
+
+      private void DecodeAll(ByteBuffer buf, IProtocolDecoderOutput output)
+      {
+         for ( ; ; )
+         {
+            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;
+            }
+         }
+      }
+
+      /// <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;
+      }
+
+      private ByteBuffer AllocateBuffer()
+      {
+         ByteBuffer buffer = ByteBuffer.Allocate(4096);
+         buffer.IsAutoExpand = true;
+         return buffer;
+      }
+   }
+}