Adding Federated Exchange Parser tests

- Validated tests pass for unit and integration tests

src/Spring.Messaging.Amqp.Rabbit/Support/ConcurrentQueueExtensions.cs

@@ -172,6 +172,17 @@ public static T Poll<T>(this BlockingCollection<T> collection, TimeSpan duration
             return default(TValue);
         }
 
+        public static void AddOrUpdate<TKey, TValue>(this IDictionary<TKey, TValue> collection, TKey key, TValue value)
+        {
+            if (collection.ContainsKey(key))
+            {
+                collection[key] = value;
+                return;
+            }
+
+            collection.Add(key, value);
+        }
+
         internal static TimeSpan Cap(TimeSpan waitTime) { return waitTime > MaxValue ? MaxValue : waitTime; }
 
         /// <summary>Lock the stack trace information of the given <paramref name="exception"/>

test/Spring.Messaging.Amqp.Rabbit.Tests/Config/FederatedExchangeParserIntegrationTests-context.xml

@@ -1,39 +1,40 @@
 <?xml version="1.0" encoding="UTF-8"?>
 
 <objects xmlns="http://www.springframework.net" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.net/schema/rabbit" xmlns:objects="http://www.springframework.net">
-    <federated-exchange name="fedDirectTest" backing-type="direct" upstream-set="upstream-set">
+
+    <federated-exchange id="fedDirectTest" name="fedDirectTest" backing-type="direct" upstream-set="upstream-set" xmlns="http://www.springframework.net/schema/rabbit">
         <direct-bindings>
             <binding queue="bucket" />
         </direct-bindings>
     </federated-exchange>
 
-    <federated-exchange name="fedTopicTest" backing-type="topic" upstream-set="upstream-set">
+    <federated-exchange id="fedTopicTest" name="fedTopicTest" backing-type="topic" upstream-set="upstream-set" xmlns="http://www.springframework.net/schema/rabbit">
         <topic-bindings>
             <binding queue="bucket" pattern="bucket.#"/>
         </topic-bindings>
     </federated-exchange>
 
-    <federated-exchange name="fedFanoutTest" backing-type="fanout" upstream-set="upstream-set">
+    <federated-exchange id="fedFanoutTest" name="fedFanoutTest" backing-type="fanout" upstream-set="upstream-set" xmlns="http://www.springframework.net/schema/rabbit">
         <fanout-bindings>
             <binding queue="bucket" />
         </fanout-bindings>
     </federated-exchange>
 
-    <federated-exchange name="fedHeadersTest" backing-type="headers" upstream-set="upstream-set">
+    <federated-exchange id="fedHeadersTest" name="fedHeadersTest" backing-type="headers" upstream-set="upstream-set" xmlns="http://www.springframework.net/schema/rabbit">
         <headers-bindings>
             <binding queue="bucket" key="type" value="bucket"  />
         </headers-bindings>
     </federated-exchange>
 
-    <fanout-exchange name="fanoutTest">
+    <fanout-exchange id="fanoutTest" name="fanoutTest" xmlns="http://www.springframework.net/schema/rabbit">
         <bindings>
             <binding queue="bucket" />
         </bindings>
     </fanout-exchange>
 
-    <queue name="bucket" />
+    <queue id="bucket" name="bucket"  xmlns="http://www.springframework.net/schema/rabbit" />
 
-    <admin id="admin-test" connection-factory="connectionFactory" auto-startup="true"/>
+    <admin id="admin" connection-factory="connectionFactory" auto-startup="true" xmlns="http://www.springframework.net/schema/rabbit" />
 
-    <connection-factory id="connectionFactory"/>
+    <connection-factory id="connectionFactory" xmlns="http://www.springframework.net/schema/rabbit" />
 </objects>

test/Spring.Messaging.Amqp.Rabbit.Tests/Config/FederatedExchangeParserIntegrationTests.cs

@@ -0,0 +1,123 @@
+// --------------------------------------------------------------------------------------------------------------------
+// <copyright file="FederatedExchangeParserIntegrationTests.cs" company="The original author or authors.">
+//   Copyright 2002-2012 the original author or authors.
+//   
+//   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.
+// </copyright>
+// --------------------------------------------------------------------------------------------------------------------
+
+#region Using Directives
+using System;
+using System.Threading;
+using Common.Logging;
+using NUnit.Framework;
+using Spring.Messaging.Amqp.Core;
+using Spring.Messaging.Amqp.Rabbit.Admin;
+using Spring.Messaging.Amqp.Rabbit.Config;
+using Spring.Messaging.Amqp.Rabbit.Connection;
+using Spring.Messaging.Amqp.Rabbit.Core;
+using Spring.Messaging.Amqp.Rabbit.Tests.Test;
+using Spring.Objects.Factory.Xml;
+using Spring.Testing.NUnit;
+#endregion
+
+namespace Spring.Messaging.Amqp.Rabbit.Tests.Config
+{
+    /// <summary>
+    /// Federated Exchange Parser Integration Tests
+    /// </summary>
+    [TestFixture]
+    [Category(TestCategory.Integration)]
+    public class FederatedExchangeParserIntegrationTests : AbstractDependencyInjectionSpringContextTests
+    {
+        public static readonly ILog Logger = LogManager.GetCurrentClassLogger();
+
+        private static readonly EnvironmentAvailable Environment = new EnvironmentAvailable("BROKER_INTEGRATION_TEST");
+
+        // @Rule
+        private readonly BrokerFederated brokerFederated = BrokerFederated.IsRunning();
+
+        // @Autowired
+        protected IConnectionFactory connectionFactory;
+
+        // @Autowired
+        protected IExchange fanoutTest;
+
+        // @Autowired
+        // @Qualifier("bucket")
+        protected Queue bucket;
+
+        // @Autowired
+        protected RabbitAdmin admin;
+
+        private RabbitBrokerAdmin brokerAdmin;
+
+        /// <summary>Initializes a new instance of the <see cref="FederatedExchangeParserIntegrationTests"/> class.</summary>
+        public FederatedExchangeParserIntegrationTests() { this.PopulateProtectedVariables = true; }
+
+        /// <summary>Gets the config locations.</summary>
+        protected override string[] ConfigLocations
+        {
+            get
+            {
+                var resourceName =
+                    @"assembly://Spring.Messaging.Amqp.Rabbit.Tests/Spring.Messaging.Amqp.Rabbit.Tests.Config/"
+                    + typeof(FederatedExchangeParserIntegrationTests).Name + "-context.xml";
+                return new[] { resourceName };
+            }
+        }
+
+        /// <summary>
+        /// Ensures that RabbitMQ is running.
+        /// </summary>
+        [TestFixtureSetUp]
+        public void FixtureSetUp()
+        {
+            NamespaceParserRegistry.RegisterParser(typeof(RabbitNamespaceHandler));
+            try
+            {
+                if (Environment.IsActive())
+                {
+                    // Set up broker admin for non-root user
+                    this.brokerAdmin = BrokerTestUtils.GetRabbitBrokerAdmin(); // "rabbit@LOCALHOST", 5672);
+                    this.brokerAdmin.StartNode();
+                }
+            }
+            catch (Exception ex)
+            {
+                Logger.Error("An error occurred during SetUp", ex);
+                Assert.Fail("An error occurred during SetUp.");
+            }
+
+            if (!this.brokerFederated.Apply())
+            {
+                Assert.Ignore("Rabbit broker is not running. Ignoring integration test fixture.");
+            }
+        }
+
+        /// <summary>The test bindings declared.</summary>
+        [Test]
+        public void TestBindingsDeclared()
+        {
+            var template = new RabbitTemplate(this.connectionFactory);
+            template.ConvertAndSend(this.fanoutTest.Name, string.Empty, "message");
+            Thread.Sleep(200);
+
+            // The queue is anonymous so it will be deleted at the end of the test, but it should get the message as long as
+            // we use the same connection
+            var result = (string)template.ReceiveAndConvert(this.bucket.Name);
+            Assert.AreEqual("message", result);
+            this.admin.DeleteExchange("fedDirectTest");
+            this.admin.DeleteExchange("fedTopicTest");
+            this.admin.DeleteExchange("fedFanoutTest");
+            this.admin.DeleteExchange("fedHeadersTest");
+        }
+    }
+}

test/Spring.Messaging.Amqp.Rabbit.Tests/Spring.Messaging.Amqp.Rabbit.Tests.2010.csproj

@@ -106,6 +106,7 @@
     <Compile Include="Config\ConnectionFactoryParserTests.cs" />
     <Compile Include="Config\ExchangeParserIntegrationTests.cs" />
     <Compile Include="Config\ExchangeParserTests.cs" />
+    <Compile Include="Config\FederatedExchangeParserIntegrationTests.cs" />
     <Compile Include="Config\ListenerContainerParserTests.cs" />
     <Compile Include="Config\ListenerContainerPlaceholderParserTests.cs" />
     <Compile Include="Config\QueueParserIntegrationTests.cs" />
@@ -146,6 +147,7 @@
     <Compile Include="Listener\SimpleMessageListenerContainerTests.cs" />
     <Compile Include="Listener\UnackedRawIntegrationTests.cs" />
     <Compile Include="Support\MoqExtensions.cs" />
+    <Compile Include="Test\BrokerFederated.cs" />
     <Compile Include="Test\BrokerPanic.cs" />
     <Compile Include="Test\BrokerRunning.cs" />
     <Compile Include="Test\BrokerTestUtils.cs" />

test/Spring.Messaging.Amqp.Rabbit.Tests/Test/BrokerFederated.cs

@@ -0,0 +1,151 @@
+// --------------------------------------------------------------------------------------------------------------------
+// <copyright file="BrokerFederated.cs" company="The original author or authors.">
+//   Copyright 2002-2012 the original author or authors.
+//   
+//   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.
+// </copyright>
+// --------------------------------------------------------------------------------------------------------------------
+
+#region Using Directives
+using System;
+using System.Collections.Generic;
+using Common.Logging;
+using NUnit.Framework;
+using Spring.Messaging.Amqp.Core;
+using Spring.Messaging.Amqp.Rabbit.Connection;
+using Spring.Messaging.Amqp.Rabbit.Core;
+using Spring.Messaging.Amqp.Rabbit.Support;
+#endregion
+
+namespace Spring.Messaging.Amqp.Rabbit.Tests.Test
+{
+    /// <summary>
+    /// TODO: Update summary.
+    /// </summary>
+    public class BrokerFederated
+    {
+        private static readonly ILog Logger = LogManager.GetCurrentClassLogger();
+
+        // Static so that we only test once on failure: speeds up test suite
+        private static readonly IDictionary<int, bool> BrokerOnline = new Dictionary<int, bool>();
+
+        // Static so that we only test once on failure
+        private static readonly IDictionary<int, bool> BrokerOffline = new Dictionary<int, bool>();
+
+        private readonly bool assumeOnline;
+
+        private readonly int DEFAULT_PORT = BrokerTestUtils.GetPort();
+
+        private int port;
+
+        private string hostName = string.Empty;
+
+        /**
+         * @return a new rule that assumes an existing running broker
+         */
+
+        /// <summary>The is running.</summary>
+        /// <returns>The Spring.Messaging.Amqp.Rabbit.Tests.Test.BrokerFederated.</returns>
+        public static BrokerFederated IsRunning() { return new BrokerFederated(); }
+
+        private BrokerFederated()
+        {
+            this.assumeOnline = true;
+            Port = this.DEFAULT_PORT;
+        }
+
+        /**
+         * @param port the port to set
+         */
+
+        /// <summary>Sets the port.</summary>
+        public int Port
+        {
+            set
+            {
+                this.port = value;
+                if (!BrokerOffline.ContainsKey(this.port))
+                {
+                    BrokerOffline.AddOrUpdate(this.port, true);
+                }
+
+                if (!BrokerOnline.ContainsKey(this.port))
+                {
+                    BrokerOnline.AddOrUpdate(this.port, true);
+                }
+            }
+        }
+
+        /**
+         * @param hostName the hostName to set
+         */
+
+        /// <summary>Sets the host name.</summary>
+        public string HostName { set { this.hostName = value; } }
+
+        /// <summary>The apply.</summary>
+        /// <returns>The System.Boolean.</returns>
+        public bool Apply()
+        {
+            // Check at the beginning, so this can be used as a static field
+            if (this.assumeOnline)
+            {
+                Assume.That(BrokerOnline.Get(this.port));
+            }
+            else
+            {
+                Assume.That(BrokerOffline.Get(this.port));
+            }
+
+            var connectionFactory = new CachingConnectionFactory();
+
+            try
+            {
+                connectionFactory.Port = this.port;
+                if (!string.IsNullOrWhiteSpace(this.hostName))
+                {
+                    connectionFactory.Host = this.hostName;
+                }
+
+                var admin = new RabbitAdmin(connectionFactory);
+                var exchange = new FederatedExchange("fedDirectRuleTest");
+                exchange.BackingType = "direct";
+                exchange.UpstreamSet = "upstream-set";
+                admin.DeclareExchange(exchange);
+                admin.DeleteExchange("fedDirectRuleTest");
+
+                BrokerOffline.AddOrUpdate(this.port, false);
+
+                if (!this.assumeOnline)
+                {
+                    Assume.That(BrokerOffline.Get(this.port));
+                }
+            }
+            catch (Exception e)
+            {
+                Logger.Warn(m => m("Not executing tests because federated connectivity test failed"), e);
+                BrokerOnline.AddOrUpdate(this.port, false);
+                if (this.assumeOnline)
+                {
+                    Assume.That(e == null, "An exception occurred.");
+                    return false;
+                }
+            }
+            finally
+            {
+                connectionFactory.Dispose();
+            }
+
+            return true;
+
+            // return super.apply(base, method, target);
+        }
+    }
+}