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

4 files changed, 109 insertions, 46 deletions

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

    }

 }