Merged revisions 539783-539788 via svnmerge from

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

........
  r539783 | tomasr | 2007-05-19 18:40:32 +0100 (Sat, 19 May 2007) | 8 lines
  
  * QPID-495 (Contributed by Carlos Medina) Implement default timeouts for AttainState and SyncWrite
  * Fix method signatures
  * Remove SSL test with client-side certificates (requires extra setup)
  * Add locks AMSQtateManager and AMQProtocolListener to prevent modification of listener collections while processing notifications
  * Add library/runtime information to ConnectionStartMethodHandler
  * Fix some compiler warnings
  * Added XML documentation for some api interfaces
........
  r539788 | tomasr | 2007-05-19 19:55:33 +0100 (Sat, 19 May 2007) | 1 line
  
  * Excluded failover tests from nant builds and SSL tests on mono
........


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

10 files changed, 217 insertions, 67 deletions

diff --git a/dotnet/Qpid.Client/Client/BasicMessageProducer.cs b/dotnet/Qpid.Client/Client/BasicMessageProducer.cs
index fd430694df..c1af826102 100644
--- a/dotnet/Qpid.Client/Client/BasicMessageProducer.cs
+++ b/dotnet/Qpid.Client/Client/BasicMessageProducer.cs
@@ -44,7 +44,7 @@ namespace Qpid.Client
 
       /// <summary>
       /// Time to live of messages. Specified in milliseconds but AMQ has 1 second resolution.
-      ///
+      /// </summary>
       private long _timeToLive;
 
       /// <summary>
@@ -88,17 +88,6 @@ namespace Qpid.Client
       /// </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,
@@ -206,15 +195,31 @@ namespace Qpid.Client
       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);
+         SendImpl(
+            _exchangeName, 
+            _routingKey, 
+            (AbstractQmsMessage)msg, 
+            deliveryMode, 
+            priority, 
+            (uint)timeToLive, 
+            _mandatory,
+            _immediate
+            );
       }
 
       public void Send(IMessage msg)
       {
          CheckNotClosed();
-         SendImpl(_exchangeName, _routingKey, (AbstractQmsMessage)msg, _deliveryMode, _messagePriority, (uint)_timeToLive,
-                  DEFAULT_MANDATORY, DEFAULT_IMMEDIATE);
+         SendImpl(
+            _exchangeName, 
+            _routingKey, 
+            (AbstractQmsMessage)msg, 
+            _deliveryMode, 
+            _messagePriority, 
+            (uint)_timeToLive,
+            _mandatory, 
+            _immediate
+            );
       }
 
       // This is a short-term hack (knowing that this code will be re-vamped sometime soon)
@@ -222,8 +227,16 @@ namespace Qpid.Client
       public void Send(IMessage msg, bool mandatory)
       {
          CheckNotClosed();
-         SendImpl(_exchangeName, _routingKey, (AbstractQmsMessage)msg, _deliveryMode, _messagePriority, (uint)_timeToLive,
-                  mandatory, DEFAULT_IMMEDIATE);
+         SendImpl(
+            _exchangeName, 
+            _routingKey, 
+            (AbstractQmsMessage)msg, 
+            _deliveryMode, 
+            _messagePriority, 
+            (uint)_timeToLive,
+            mandatory, 
+            _immediate
+            );
       }
 
       public long TimeToLive
@@ -248,6 +261,11 @@ namespace Qpid.Client
 
       public string MimeType
       {
+         get
+         {
+            CheckNotClosed();
+            return _mimeType;
+         }
          set
          {
             CheckNotClosed();
@@ -257,6 +275,11 @@ namespace Qpid.Client
 
       public string Encoding
       {
+         get
+         {
+            CheckNotClosed();
+            return _encoding;
+         }
          set
          {
             CheckNotClosed();
diff --git a/dotnet/Qpid.Client/Client/Handler/ConnectionStartMethodHandler.cs b/dotnet/Qpid.Client/Client/Handler/ConnectionStartMethodHandler.cs
index 99ee7e2587..3cb7c76e23 100644
--- a/dotnet/Qpid.Client/Client/Handler/ConnectionStartMethodHandler.cs
+++ b/dotnet/Qpid.Client/Client/Handler/ConnectionStartMethodHandler.cs
@@ -89,16 +89,16 @@ namespace Qpid.Client.Handler
 
         private string GetFullSystemInfo()
         {
-            /*StringBuffer fullSystemInfo = new StringBuffer();
-            fullSystemInfo.append(System.getProperty("java.runtime.name"));
-            fullSystemInfo.append(", " + System.getProperty("java.runtime.version"));
-            fullSystemInfo.append(", " + System.getProperty("java.vendor"));
-            fullSystemInfo.append(", " + System.getProperty("os.arch"));
-            fullSystemInfo.append(", " + System.getProperty("os.name"));
-            fullSystemInfo.append(", " + System.getProperty("os.version"));
-            fullSystemInfo.append(", " + System.getProperty("sun.os.patch.level"));*/
-            // TODO: add in details here
-            return ".NET 1.1 Client";
+            StringBuilder sysInfo = new StringBuilder();
+            // check if we're running on mono or .net
+            Type monoRuntime = Type.GetType("Mono.Runtime");
+            if ( monoRuntime != null )
+               sysInfo.Append("Mono");
+            else
+               sysInfo.Append(".NET");
+            sysInfo.Append(" ").Append(Environment.Version);
+            sysInfo.Append(", ").Append(Environment.OSVersion);
+            return sysInfo.ToString();
         }
 
        private string ChooseMechanism(string mechanisms)
diff --git a/dotnet/Qpid.Client/Client/Message/AbstractQmsMessage.cs b/dotnet/Qpid.Client/Client/Message/AbstractQmsMessage.cs
index 8e90e852dd..7a28d7a85f 100644
--- a/dotnet/Qpid.Client/Client/Message/AbstractQmsMessage.cs
+++ b/dotnet/Qpid.Client/Client/Message/AbstractQmsMessage.cs
@@ -30,8 +30,6 @@ namespace Qpid.Client.Message
 {
     public abstract class AbstractQmsMessage : AMQMessage, IMessage
     {
-        private static readonly ILog _log = LogManager.GetLogger(typeof(AbstractQmsMessage));
-
         protected bool _redelivered;
 
         protected ByteBuffer _data;
diff --git a/dotnet/Qpid.Client/Client/Protocol/AMQProtocolListener.cs b/dotnet/Qpid.Client/Client/Protocol/AMQProtocolListener.cs
index 2b6f239127..2bf9f9a4c1 100644
--- a/dotnet/Qpid.Client/Client/Protocol/AMQProtocolListener.cs
+++ b/dotnet/Qpid.Client/Client/Protocol/AMQProtocolListener.cs
@@ -234,16 +234,21 @@ namespace Qpid.Client.Protocol
         {
             // FIXME: not sure if required as StateManager is in _frameListeners. Probably something to do with fail-over.
             _stateManager.Error(e);
-
-            foreach (IAMQMethodListener listener in _frameListeners)
+            lock ( _lock )
             {
-                listener.Error(e);
+               foreach ( IAMQMethodListener listener in _frameListeners )
+               {
+                  listener.Error(e);
+               }
             }
         }
 
         public void AddFrameListener(IAMQMethodListener listener)
         {
-            _frameListeners.Add(listener);
+           lock ( _lock )
+           {
+              _frameListeners.Add(listener);
+           }
         }
 
         public void RemoveFrameListener(IAMQMethodListener listener)
@@ -252,7 +257,10 @@ namespace Qpid.Client.Protocol
             {
                 _log.Debug("Removing frame listener: " + listener.ToString());
             }
-            _frameListeners.Remove(listener);
+            lock ( _lock )
+            {
+               _frameListeners.Remove(listener);
+            }
         }
 
         public void BlockUntilNotFailingOver()
diff --git a/dotnet/Qpid.Client/Client/Protocol/DefaultTimeouts.cs b/dotnet/Qpid.Client/Client/Protocol/DefaultTimeouts.cs
new file mode 100644
index 0000000000..0638bf0704
--- /dev/null
+++ b/dotnet/Qpid.Client/Client/Protocol/DefaultTimeouts.cs
@@ -0,0 +1,47 @@
+/*

+ *

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

+

+namespace Qpid.Client.Protocol

+{

+   /// <summary>

+   /// Default timeout values for the protocol

+   /// </summary>

+   sealed class DefaultTimeouts

+   {

+      /// <summary>

+      /// Maximum number of milliseconds to wait for a state change

+      /// in the protocol's state machine

+      /// </summary>

+      public const int MaxWaitForState = 30* 1000;

+      /// <summary>

+      /// Maximum number of milliseconds to wait for a reply

+      /// frame when doing synchronous writer to the broker

+      /// </summary>

+      public const int MaxWaitForSyncWriter = 30 * 1000;

+

+      private DefaultTimeouts()

+      {

+      }

+   }

+}

diff --git a/dotnet/Qpid.Client/Client/Protocol/Listener/BlockingMethodFrameListener.cs b/dotnet/Qpid.Client/Client/Protocol/Listener/BlockingMethodFrameListener.cs
index 99643fe59f..a020efbf21 100644
--- a/dotnet/Qpid.Client/Client/Protocol/Listener/BlockingMethodFrameListener.cs
+++ b/dotnet/Qpid.Client/Client/Protocol/Listener/BlockingMethodFrameListener.cs
@@ -80,9 +80,10 @@ namespace Qpid.Client.Protocol.Listener
         /// <summary>
         /// This method is called by the thread that wants to wait for a frame.
         /// </summary>
-        public AMQMethodEvent BlockForFrame()
+        /// <param name="timeout">Set the number of milliseconds to wait</param>
+        public AMQMethodEvent BlockForFrame(int timeout)
         {
-            _resetEvent.WaitOne();
+            _resetEvent.WaitOne(timeout, true);
             //at this point the event will have been signalled. The error field might or might not be set
             // depending on whether an error occurred
             if (_error != null)
diff --git a/dotnet/Qpid.Client/Client/Protocol/ProtocolWriter.cs b/dotnet/Qpid.Client/Client/Protocol/ProtocolWriter.cs
index 32847f9b9b..0ef337501e 100644
--- a/dotnet/Qpid.Client/Client/Protocol/ProtocolWriter.cs
+++ b/dotnet/Qpid.Client/Client/Protocol/ProtocolWriter.cs
@@ -23,6 +23,8 @@ using Qpid.Client.Protocol.Listener;
 using Qpid.Client.Transport;
 using Qpid.Framing;
 
+using log4net;
+
 namespace Qpid.Client.Protocol
 {
     /// <summary>
@@ -30,6 +32,9 @@ namespace Qpid.Client.Protocol
     /// </summary>
     public class ProtocolWriter
     {
+
+        private ILog _logger = LogManager.GetLogger(typeof(ProtocolWriter));
+
         IProtocolWriter _protocolWriter;
         IProtocolListener _protocolListener;
         
@@ -51,13 +56,15 @@ namespace Qpid.Client.Protocol
         /// </summary>
         /// <param name="frame">the frame</param>
         /// <param name="listener">the blocking listener. Note the calling thread will block.</param>         
-        private AMQMethodEvent SyncWrite(AMQFrame frame, BlockingMethodFrameListener listener)            
+        /// <param name="timeout">set the number of milliseconds to wait</param>
+        private AMQMethodEvent SyncWrite(AMQFrame frame, BlockingMethodFrameListener listener, int timeout)
         {
             try
             {
                 _protocolListener.AddFrameListener(listener);
                 _protocolWriter.Write(frame);
-                return listener.BlockForFrame();                
+                
+                return listener.BlockForFrame(timeout);
             }
             finally
             {
@@ -67,11 +74,32 @@ namespace Qpid.Client.Protocol
             // that matches the criteria defined in the blocking listener
         }
 
+        /// <summary>
+        /// Convenience method that writes a frame to the protocol session and waits for
+        /// a particular response. Equivalent to calling getProtocolSession().write() then
+        /// waiting for the response.
+        /// </summary>
+        /// <param name="frame">the frame</param>
+        /// <param name="responseType">the type of method response</param>
         public AMQMethodEvent SyncWrite(AMQFrame frame, Type responseType)
         {
             // TODO: If each frame knew it's response type, then the responseType argument would
             // TODO: not be neccesary.
-            return SyncWrite(frame, new SpecificMethodFrameListener(frame.Channel, responseType));
+            return SyncWrite(frame, responseType, DefaultTimeouts.MaxWaitForSyncWriter);
+        }
+
+        /// <summary>
+        /// Convenience method that writes a frame to the protocol session and waits for
+        /// a particular response. Equivalent to calling getProtocolSession().write() then
+        /// waiting for the response.
+        /// </summary>
+        /// <param name="frame">the frame</param>
+        /// <param name="responseType">the type of method response</param>
+        /// <param name="timeout">set the number of milliseconds to wait</param>
+        /// <returns>set the number of milliseconds to wait</returns>
+        public AMQMethodEvent SyncWrite(AMQFrame frame, Type responseType, int timeout)
+        {
+            return SyncWrite(frame, new SpecificMethodFrameListener(frame.Channel, responseType), timeout);
         }
     }
 }
diff --git a/dotnet/Qpid.Client/Client/State/AMQStateManager.cs b/dotnet/Qpid.Client/Client/State/AMQStateManager.cs
index 1233f9d836..9ce6d3c76a 100644
--- a/dotnet/Qpid.Client/Client/State/AMQStateManager.cs
+++ b/dotnet/Qpid.Client/Client/State/AMQStateManager.cs
@@ -43,13 +43,15 @@ namespace Qpid.Client.State
         /// Maps from an AMQState instance to a Map from Class to StateTransitionHandler.
         /// The class must be a subclass of AMQFrame.
         /// </summary>
-        private readonly IDictionary _state2HandlersMap = new Hashtable();
-
-        //private CopyOnWriteArraySet _stateListeners = new CopyOnWriteArraySet();
-        private ArrayList _stateListeners = ArrayList.Synchronized(new ArrayList(5));
+        private readonly IDictionary _state2HandlersMap;
+        private ArrayList _stateListeners;
+        private object _syncLock;
         
         public AMQStateManager()
         {
+            _syncLock = new object();
+            _state2HandlersMap = new Hashtable();
+            _stateListeners = ArrayList.Synchronized(new ArrayList(5));
             _currentState = AMQState.CONNECTION_NOT_STARTED;
             RegisterListeners();
         }
@@ -132,18 +134,24 @@ namespace Qpid.Client.State
             AMQState oldState = _currentState;
             _currentState = newState;
 
-            foreach (IStateListener l in _stateListeners)
+            lock ( _syncLock )
             {
-                l.StateChanged(oldState, newState);
+               foreach ( IStateListener l in _stateListeners )
+               {
+                  l.StateChanged(oldState, newState);
+               }
             }
         }
 
         public void Error(Exception e)
         {
             _logger.Debug("State manager receive error notification: " + e);
-            foreach (IStateListener l in _stateListeners)
+            lock ( _syncLock )
             {
-                l.Error(e);
+               foreach ( IStateListener l in _stateListeners )
+               {
+                  l.Error(e);
+               }
             }
         }
 
@@ -206,23 +214,37 @@ namespace Qpid.Client.State
         public void AddStateListener(IStateListener listener)
         {
             _logger.Debug("Adding state listener");
-            _stateListeners.Add(listener);
+            lock ( _syncLock )
+            {
+               _stateListeners.Add(listener);
+            }
         }
 
         public void RemoveStateListener(IStateListener listener)
         {
-            _stateListeners.Remove(listener);
+           lock ( _syncLock )
+           {
+              _stateListeners.Remove(listener);
+           }
         }
 
         public void AttainState(AMQState s)
         {
             if (_currentState != s)
             {
-                _logger.Debug("Adding state wait to reach state " + s);
-                StateWaiter sw = new StateWaiter(s);
-                AddStateListener(sw);
-                sw.WaituntilStateHasChanged();
-                // at this point the state will have changed.
+                StateWaiter sw = null;
+                try
+                {
+                    _logger.Debug("Adding state wait to reach state " + s);
+                    sw = new StateWaiter(s);
+                    AddStateListener(sw);
+                    sw.WaituntilStateHasChanged();
+                    // at this point the state will have changed.
+                }
+                finally
+                { 
+                    RemoveStateListener(sw);
+                }
             }
         }        
     }
diff --git a/dotnet/Qpid.Client/Client/State/StateWaiter.cs b/dotnet/Qpid.Client/Client/State/StateWaiter.cs
index cb7f604499..34667da744 100644
--- a/dotnet/Qpid.Client/Client/State/StateWaiter.cs
+++ b/dotnet/Qpid.Client/Client/State/StateWaiter.cs
@@ -20,6 +20,7 @@
  */
 using System;
 using System.Threading;
+using Qpid.Client.Protocol;
 using log4net;
 
 namespace Qpid.Client.State
@@ -29,6 +30,7 @@ namespace Qpid.Client.State
         private static readonly ILog _logger = LogManager.GetLogger(typeof(StateWaiter));
 
         private readonly AMQState _state;
+        private AMQState _newState;
 
         private volatile bool _newStateAchieved;
 
@@ -42,7 +44,8 @@ namespace Qpid.Client.State
         }
 
         public void StateChanged(AMQState oldState, AMQState newState)
-        {            
+        {
+            _newState = newState;
             if (_logger.IsDebugEnabled)
             {
                 _logger.Debug("stateChanged called");
@@ -76,23 +79,42 @@ namespace Qpid.Client.State
             // The guard is required in case we are woken up by a spurious
             // notify().
             //
-            while (!_newStateAchieved && _exception == null)
-            {                    
+
+            TimeSpan waitTime = TimeSpan.FromMilliseconds(DefaultTimeouts.MaxWaitForState);
+            DateTime waitUntilTime = DateTime.Now + waitTime;
+
+            while ( !_newStateAchieved 
+               && _exception == null 
+               && waitTime.TotalMilliseconds > 0 )
+            {
                 _logger.Debug("State not achieved so waiting...");
-                _resetEvent.WaitOne();                    
+                try
+                {
+                    _resetEvent.WaitOne(waitTime, true);
+                }
+                finally
+                {
+                    if (!_newStateAchieved)
+                    {
+                        waitTime = waitUntilTime - DateTime.Now;
+                    }
+                }
             }
 
             if (_exception != null)
             {
                 _logger.Debug("Throwable reached state waiter: " + _exception);
                 if (_exception is AMQException)
-                {
                     throw _exception;
-                }
                 else
-                {
                     throw new AMQException("Error: "  + _exception, _exception);
-                }
+            }
+
+            if (!_newStateAchieved)
+            {
+                string error = string.Format("State not achieved within permitted time. Current state: {0}, desired state: {1}", _state, _newState);
+                _logger.Warn(error);
+                throw new AMQException(error);
             }
         }
     }
diff --git a/dotnet/Qpid.Client/Client/Transport/AmqpChannel.cs b/dotnet/Qpid.Client/Client/Transport/AmqpChannel.cs
index 4e4ca03322..e23037d1de 100644
--- a/dotnet/Qpid.Client/Client/Transport/AmqpChannel.cs
+++ b/dotnet/Qpid.Client/Client/Transport/AmqpChannel.cs
@@ -82,10 +82,11 @@ namespace Qpid.Client.Transport
             _byteChannel.Write(Encode(o));
         }
 
-        private void OnAsyncWriteDone(IAsyncResult result)
-        {
-           _byteChannel.EndWrite(result);
-        }
+        // not used for now
+        //private void OnAsyncWriteDone(IAsyncResult result)
+        //{
+        //   _byteChannel.EndWrite(result);
+        //}
 
         private void Decode(ByteBuffer buffer)
         {