Merge bug25082 into default

Author: Tim Watson

projects/client/RabbitMQ.Client/src/client/api/IModel.cs

@@ -126,11 +126,15 @@ public interface IModel: IDisposable
         ///otherwise.</summary>
         ShutdownEventArgs CloseReason { get; }
 
-        ///<summary>Returns true if the session is still in a state
+        ///<summary>Returns true if the model is still in a state
         ///where it can be used. Identical to checking if CloseReason
         ///== null.</summary>
         bool IsOpen { get; }
 
+        ///<summary>Returns true if the model is no longer in a state
+        ///where it can be used.</summary>
+        bool IsClosed { get; }
+
         ///<summary>When in confirm mode, return the sequence number
         ///of the next message to be published.</summary>
         ulong NextPublishSeqNo { get; }

projects/client/RabbitMQ.Client/src/client/api/SslHelper.cs

@@ -90,9 +90,15 @@ private bool CertificateValidationCallback(object sender,
         public static Stream TcpUpgrade(Stream tcpStream, SslOption sslOption, int timeout)
         {
             SslHelper helper = new SslHelper(sslOption);
+
+            RemoteCertificateValidationCallback remoteCertValidator =
+              sslOption.CertificateValidationCallback ?? new RemoteCertificateValidationCallback(helper.CertificateValidationCallback);
+            LocalCertificateSelectionCallback localCertSelector =
+              sslOption.CertificateSelectionCallback ?? new LocalCertificateSelectionCallback(helper.CertificateSelectionCallback);
+
             SslStream sslStream = new SslStream(tcpStream, false,
-                                                new RemoteCertificateValidationCallback(helper.CertificateValidationCallback),
-                                                new LocalCertificateSelectionCallback(helper.CertificateSelectionCallback));
+                                                remoteCertValidator,
+                                                localCertSelector);
 
             sslStream.AuthenticateAsClient(sslOption.ServerName,
                                            sslOption.Certs,

projects/client/RabbitMQ.Client/src/client/api/SslOption.cs

@@ -136,6 +136,18 @@ public SslPolicyErrors AcceptablePolicyErrors
             set { m_acceptablePolicyErrors = value; }
         }
 
+        /// <summary>
+        /// An optional client specified SSL certificate validation callback.  If this is not specified,
+        /// the default callback will be used in conjunction with the AcceptablePolicyErrors property to 
+        /// determine if the remote server certificate is valid.
+        /// </summary>
+        public RemoteCertificateValidationCallback CertificateValidationCallback { get; set; }
+
+        /// <summary>
+        /// An optional client specified SSL certificate selection callback.  If this is not specified,
+        /// the first valid certificate found will be used.
+        /// </summary>
+        public LocalCertificateSelectionCallback CertificateSelectionCallback { get; set; }
 
         ///<summary>Construct an SslOption specifying both the server cannonical name
         ///and the client's certificate path.
@@ -145,6 +157,8 @@ public SslOption(string serverName, string certPath, bool enabled)
             m_serverName= serverName;
             m_certPath = certPath;
             m_enabled = enabled;
+            CertificateValidationCallback = null;
+            CertificateSelectionCallback  = null;
         }
 
         ///<summary>Construct an SslOption with just the server cannonical name.

projects/client/RabbitMQ.Client/src/client/exceptions/AlreadyClosedException.cs

@@ -49,6 +49,6 @@ public class AlreadyClosedException: OperationInterruptedException
         ///<summary>Construct an instance containing the given
         ///shutdown reason.</summary>
         public AlreadyClosedException(ShutdownEventArgs reason)
-            : base(reason) { }
+            : base(reason, "Already closed") { }
     }
 }

projects/client/RabbitMQ.Client/src/client/exceptions/OperationInterruptedException.cs

@@ -81,6 +81,16 @@ public OperationInterruptedException(ShutdownEventArgs reason)
             m_shutdownReason = reason;
         }
 
+        ///<summary>Construct an OperationInterruptedException with
+        ///the passed-in explanation and prefix, if any.</summary>
+        public OperationInterruptedException(ShutdownEventArgs reason, String prefix)
+        : base(reason == null ? (prefix + ": The AMQP operation was interrupted") :
+                   string.Format("{0}: The AMQP operation was interrupted: {1}",
+                                 prefix, reason))
+        {
+            m_shutdownReason = reason;
+        }
+
         ///<summary>Retrieves the explanation for the shutdown. May
         ///return null if no explanation is available.</summary>
         public ShutdownEventArgs ShutdownReason { get { return m_shutdownReason; } }

projects/client/RabbitMQ.Client/src/client/impl/ConnectionBase.cs

@@ -485,6 +485,11 @@ public void Close(ShutdownEventArgs reason, bool abort, int timeout)
                     if (!abort)
                         throw ace;
                 }
+                catch (NotSupportedException nse)
+                {
+                    // buffered stream had unread data in it and Flush()
+                    // was called, ignore to not confuse the user
+                }
                 catch (IOException ioe)
                 {
                     if (m_model0.CloseReason == null)
@@ -665,7 +670,15 @@ public void MainLoop()
                 // connection closes.
                 if (shutdownCleanly)
                 {
-                    ClosingLoop();
+                    try
+                    {
+                        ClosingLoop();
+                    } catch (SocketException se)
+                    {
+                        // means that socket was closed when frame handler
+                        // attempted to use it. Since we are shutting down,
+                        // ignore it.
+                    }
                 }
 
                 FinishClose();
@@ -1055,9 +1068,7 @@ protected void StartAndTune()
                 connectionStartCell.Value;
 
             if (connectionStart == null){
-                throw new ProtocolVersionMismatchException(Protocol.MajorVersion,
-                                                           Protocol.MinorVersion,
-                                                           -1, -1);
+                throw new IOException("connection.start was never received, likely due to a network timeout");
             }
 
             ServerProperties = connectionStart.m_serverProperties;

projects/client/RabbitMQ.Client/src/client/impl/ModelBase.cs

@@ -494,16 +494,7 @@ public void Enqueue(IRpcContinuation k)
         public void TransmitAndEnqueue(Command cmd, IRpcContinuation k)
         {
             Enqueue(k);
-            try
-            {
-                m_session.Transmit(cmd);
-            }
-            catch (AlreadyClosedException)
-            {
-                // Ignored, since the continuation will be told about
-                // the closure via an OperationInterruptedException because
-                // of the shutdown event propagation.
-            }
+            m_session.Transmit(cmd);
         }
 
         public ShutdownEventArgs CloseReason
@@ -522,6 +513,15 @@ public bool IsOpen
             }
         }
 
+        public bool IsClosed
+        {
+            get
+            {
+                return !IsOpen;
+            }
+        }
+
+
         public ulong NextPublishSeqNo
         {
             get
@@ -894,16 +894,7 @@ private QueueDeclareOk QueueDeclare(string queue, bool passive, bool durable, bo
         {
             QueueDeclareRpcContinuation k = new QueueDeclareRpcContinuation();
             Enqueue(k);
-            try
-            {
-                _Private_QueueDeclare(queue, passive, durable, exclusive, autoDelete, false, arguments);
-            }
-            catch (AlreadyClosedException)
-            {
-                // Ignored, since the continuation will be told about
-                // the closure via an OperationInterruptedException because
-                // of the shutdown event propagation.
-            }
+            _Private_QueueDeclare(queue, passive, durable, exclusive, autoDelete, false, arguments);
             k.GetReply();
             return k.m_result;
         }
@@ -988,7 +979,7 @@ public bool WaitForConfirms(TimeSpan timeout, out bool timedOut)
             {
                 while (true)
                 {
-                    if (CloseReason != null)
+                    if (!IsOpen)
                         throw new AlreadyClosedException(CloseReason);
 
                     if (m_unconfirmedSet.Count == 0)
@@ -1095,17 +1086,8 @@ public string BasicConsume(string queue,
             Enqueue(k);
             // Non-nowait. We have an unconventional means of getting
             // the RPC response, but a response is still expected.
-            try
-            {
-                _Private_BasicConsume(queue, consumerTag, noLocal, noAck, exclusive,
+            _Private_BasicConsume(queue, consumerTag, noLocal, noAck, exclusive,
                     /*nowait:*/ false, arguments);
-            }
-            catch (AlreadyClosedException)
-            {
-                // Ignored, since the continuation will be told about
-                // the closure via an OperationInterruptedException because
-                // of the shutdown event propagation.
-            }
             k.GetReply();
             string actualConsumerTag = k.m_consumerTag;
 
@@ -1141,17 +1123,7 @@ public void BasicCancel(string consumerTag)
 
             Enqueue(k);
 
-            try
-            {
-                _Private_BasicCancel(consumerTag, false);
-            }
-            catch (AlreadyClosedException)
-            {
-                // Ignored, since the continuation will be told about
-                // the closure via an OperationInterruptedException because
-                // of the shutdown event propagation.
-            }
-
+            _Private_BasicCancel(consumerTag, false);
             k.GetReply();
 
             ModelShutdown -= new ModelShutdownEventHandler(k.m_consumer.HandleModelShutdown);
@@ -1197,16 +1169,7 @@ public BasicGetResult BasicGet(string queue,
         {
             BasicGetRpcContinuation k = new BasicGetRpcContinuation();
             Enqueue(k);
-            try
-            {
-                _Private_BasicGet(queue, noAck);
-            }
-            catch (AlreadyClosedException)
-            {
-                // Ignored, since the continuation will be told about
-                // the closure via an OperationInterruptedException because
-                // of the shutdown event propagation.
-            }
+            _Private_BasicGet(queue, noAck);
             k.GetReply();
             return k.m_result;
         }
@@ -1218,18 +1181,7 @@ public void BasicRecover(bool requeue)
             SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation();
 
             Enqueue(k);
-
-            try
-            {
-                _Private_BasicRecover(requeue);
-            }
-            catch (AlreadyClosedException)
-            {
-                // Ignored, since the continuation will be told about
-                // the closure via an OperationInterruptedException because
-                // of the shutdown event propagation.
-            }
-
+            _Private_BasicRecover(requeue);
             k.GetReply();
         }
 
@@ -1461,7 +1413,9 @@ public ConnectionSecureOrTune ConnectionStartOk(IDictionary<string, object> clie
             }
             catch (AlreadyClosedException)
             {
-                // Ignored, see BasicGet
+                // let continuation throw OperationInterruptedException,
+                // which is a much more suitable exception before connection
+                // negotiation finishes
             }
             k.GetReply();
             return k.m_result;
@@ -1490,7 +1444,9 @@ public ConnectionSecureOrTune ConnectionSecureOk(byte[] response)
             }
             catch (AlreadyClosedException)
             {
-                // Ignored, see BasicGet
+                // let continuation throw OperationInterruptedException,
+                // which is a much more suitable exception before connection
+                // negotiation finishes
             }
             k.GetReply();
             return k.m_result;
@@ -1535,7 +1491,9 @@ public string ConnectionOpen(string virtualHost,
             }
             catch (AlreadyClosedException)
             {
-                // Ignored, see BasicGet
+                // let continuation throw OperationInterruptedException,
+                // which is a much more suitable exception before connection
+                // negotiation finishes
             }
             k.GetReply();
             if (k.m_redirect) {

projects/client/RabbitMQ.Client/src/client/messagepatterns/Subscription.cs

@@ -212,7 +212,7 @@ public BasicDeliverEventArgs Next()
                 // from under us by the operation of Close() from
                 // another thread.
                 QueueingBasicConsumer consumer = m_consumer;
-                if (consumer == null) {
+                if (consumer == null || m_model.IsClosed) {
                     // Closed!
                     m_latestEvent = null;
                 } else {
@@ -275,9 +275,11 @@ public bool Next(int millisecondsTimeout, out BasicDeliverEventArgs result)
                 // from under us by the operation of Close() from
                 // another thread.
                 QueueingBasicConsumer consumer = m_consumer;
-                if (consumer == null) {
+                if (consumer == null || m_model.IsClosed) {
                     // Closed!
                     m_latestEvent = null;
+                    result = null;
+                    return false;
                 } else {
                     BasicDeliverEventArgs qValue;
                     if (!consumer.Queue.Dequeue(millisecondsTimeout, out qValue)) {

projects/client/Unit/src/unit/TestConnectionBlocked.cs

@@ -162,7 +162,7 @@ public static bool IsRunningOnMono()
         protected void Publish(IConnection conn)
         {
             IModel ch = conn.CreateModel();
-            ch.BasicPublish("", "amq.fanout", null, enc.GetBytes("message"));
+            ch.BasicPublish("amq.fanout", "", null, enc.GetBytes("message"));
         }
 
         protected override void ReleaseResources()