Merge bug26079 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/impl/ModelBase.cs

@@ -513,6 +513,15 @@ public bool IsOpen
             }
         }
 
+        public bool IsClosed
+        {
+            get
+            {
+                return !IsOpen;
+            }
+        }
+
+
         public ulong NextPublishSeqNo
         {
             get
@@ -1404,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;
@@ -1433,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;
@@ -1478,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/TestMessagePatternsSubscription.cs

@@ -0,0 +1,79 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 1.1.
+//
+// The APL v2.0:
+//
+//---------------------------------------------------------------------------
+//   Copyright (C) 2007-2013 GoPivotal, Inc.
+//
+//   Licensed 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.
+//---------------------------------------------------------------------------
+//
+// The MPL v1.1:
+//
+//---------------------------------------------------------------------------
+//  The contents of this file are subject to the Mozilla Public License
+//  Version 1.1 (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.mozilla.org/MPL/
+//
+//  Software distributed under the License is distributed on an "AS IS"
+//  basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+//  the License for the specific language governing rights and
+//  limitations under the License.
+//
+//  The Original Code is RabbitMQ.
+//
+//  The Initial Developer of the Original Code is GoPivotal, Inc.
+//  Copyright (c) 2007-2014 GoPivotal, Inc.  All rights reserved.
+//---------------------------------------------------------------------------
+
+using NUnit.Framework;
+
+using System;
+using System.Text;
+using System.Threading;
+using System.Diagnostics;
+
+using RabbitMQ.Client.Events;
+using RabbitMQ.Client.MessagePatterns;
+
+namespace RabbitMQ.Client.Unit {
+    [TestFixture]
+    public class TestMessagePatternsSubscription : IntegrationFixture {
+        UTF8Encoding enc = new UTF8Encoding();
+
+        [Test]
+        public void TestChannelClosureIsObservableOnSubscription()
+        {
+            string q = Model.QueueDeclare();
+            Subscription sub = new Subscription(Model, q, true);
+
+            BasicDeliverEventArgs r1;
+            Assert.IsFalse(sub.Next(100, out r1));
+
+            Model.BasicPublish("", q, null, enc.GetBytes("a message"));
+            Model.BasicPublish("", q, null, enc.GetBytes("a message"));
+
+            BasicDeliverEventArgs r2;
+            Assert.IsTrue(sub.Next(1000, out r2));
+            Assert.IsNotNull(sub.Next());
+
+            Model.Close();
+            Assert.IsNull(sub.Next());
+
+            BasicDeliverEventArgs r3;
+            Assert.IsFalse(sub.Next(100, out r3));
+        }
+    }
+}