Merge pull request #36 from hansehe/feature/heh-alternate_exchanges

Feature/heh alternate exchanges
rebus-org · Nov 30, 2018 · cb3479e · cb3479e
2 parents ae95dd9 + f47a659
commit cb3479e
Show file tree

Hide file tree

Showing 3 changed files with 68 additions and 5 deletions.
diff --git a/Rebus.RabbitMq.Tests/RabbitMqCustomExchangeNamesTest.cs b/Rebus.RabbitMq.Tests/RabbitMqCustomExchangeNamesTest.cs
@@ -4,6 +4,7 @@
 using NUnit.Framework;
 using Rebus.Activation;
 using Rebus.Config;
+using Rebus.Logging;
 using Rebus.Tests.Contracts;
 using Rebus.Tests.Contracts.Extensions;
 // ReSharper disable ArgumentsStyleNamedExpression
@@ -53,5 +54,31 @@ public async Task CanUseCustomExchangeName()
             Assert.That(RabbitMqTransportFactory.ExchangeExists(customDirectExchangeName), Is.True);
             Assert.That(RabbitMqTransportFactory.ExchangeExists(customTopicExchangeName), Is.True);
         }
+
+        [Test]
+        public async Task CanUseAlternateCustomExchangeName()
+        {
+            const string connectionString = RabbitMqTransportFactory.ConnectionString;
+
+            var rabbitMqTransport = new RabbitMqTransport(connectionString, "inputQueue", new ConsoleLoggerFactory(false));
+
+            var defaultTopicExchange = "defaultTopicExchange";
+            rabbitMqTransport.AllowPublishOnAlternateExchanges();
+            rabbitMqTransport.SetTopicExchangeName(defaultTopicExchange);
+
+            var topic = "myTopic";
+            var alternateExchange = "alternateExchange";
+
+            var topicWithAlternateExchange = $"{topic}@{alternateExchange}";
+
+            var subscriberAddresses = await rabbitMqTransport.GetSubscriberAddresses(topicWithAlternateExchange);
+            Assert.That(subscriberAddresses[0], Is.EqualTo(topicWithAlternateExchange));
+
+            subscriberAddresses = await rabbitMqTransport.GetSubscriberAddresses(topic);
+            Assert.That(subscriberAddresses[0], Is.EqualTo($"{topic}@{defaultTopicExchange}"));
+
+            subscriberAddresses = await rabbitMqTransport.GetSubscriberAddresses(topic + '@');
+            Assert.That(subscriberAddresses[0], Is.EqualTo($"{topic}@@{defaultTopicExchange}"));
+        }
     }
 }
diff --git a/Rebus.RabbitMq/Config/RabbitMqOptionsBuilder.cs b/Rebus.RabbitMq/Config/RabbitMqOptionsBuilder.cs
@@ -187,11 +187,23 @@ public RabbitMqOptionsBuilder EnablePublisherConfirms(bool value = true)
             PublisherConfirms = value;
             return this;
         }
+
+        /// <summary>
+        /// Publish a topic on an alternate exchange.
+        /// Use the following syntax to publish a topic on an alternate exchange called "alternateExchange":
+        ///     "topic@alternateExchange"
+        /// </summary>
+        public RabbitMqOptionsBuilder AllowPublishOnAlternateExchanges(bool value = true)
+        {
+            PublishOnAlternateExchanges = value;
+            return this;
+        }
 
         internal bool? DeclareExchanges { get; private set; }
         internal bool? DeclareInputQueue { get; private set; }
         internal bool? BindInputQueue { get; private set; }
         internal bool? PublisherConfirms { get; private set; }
+        internal bool? PublishOnAlternateExchanges { get; private set; }
 
         internal string DirectExchangeName { get; private set; }
         internal string TopicExchangeName { get; private set; }
@@ -256,6 +268,11 @@ internal void Configure(RabbitMqTransport transport)
             {
                 transport.EnablePublisherConfirms(PublisherConfirms.Value);
             }
+
+            if (PublishOnAlternateExchanges.HasValue)
+            {
+                transport.AllowPublishOnAlternateExchanges(PublishOnAlternateExchanges.Value);
+            }
 
             transport.SetInputQueueOptions(QueueOptions);
             transport.SetExchangeOptions(ExchangeOptions);

diff --git a/Rebus.RabbitMq/RabbitMq/RabbitMqTransport.cs b/Rebus.RabbitMq/RabbitMq/RabbitMqTransport.cs
@@ -47,6 +47,7 @@ public class RabbitMqTransport : ITransport, IDisposable, IInitializable, ISubsc
         bool _declareInputQueue = true;
         bool _bindInputQueue = true;
         bool _publisherConfirmsEnabled;
+        bool _allowPublishOnAlternateExchanges;
 
         string _directExchangeName = RabbitMqOptionsBuilder.DefaultDirectExchangeName;
         string _topicExchangeName = RabbitMqOptionsBuilder.DefaultTopicExchangeName;
@@ -138,6 +139,16 @@ public void EnablePublisherConfirms(bool value = true)
         {
             _publisherConfirmsEnabled = value;
         }
+
+        /// <summary>
+        /// Publish a topic on an alternate exchange.
+        /// Use the following syntax to publish a topic on an alternate exchange called "alternateExchange":
+        ///     "topic@alternateExchange"
+        /// </summary>
+        public void AllowPublishOnAlternateExchanges(bool value = true)
+        {
+            _allowPublishOnAlternateExchanges = value;
+        }
 
         /// <summary>
         /// Sets the name of the exchange used to send point-to-point messages
@@ -526,6 +537,11 @@ async Task SendOutgoingMessages(ITransactionContext context, ConcurrentQueue<Out
                     EnsureQueueExists(routingKey, model);
                 }
 
+                if (_publisherConfirmsEnabled)
+                {
+                    model.ConfirmSelect();
+                }
+
                 model.BasicPublish(
                     exchange: exchange,
                     routingKey: routingKey.RoutingKey,
@@ -696,11 +712,6 @@ IModel GetModel(ITransactionContext context)
 
                 context.OnDisposed(() => _models.Enqueue(newModel));
 
-                if (_publisherConfirmsEnabled)
-                {
-                    newModel.ConfirmSelect();
-                }
-
                 // Configure registered events on model
                 _callbackOptions?.ConfigureEvents(newModel);
 
@@ -753,6 +764,14 @@ public void Dispose()
         /// </summary>
         public async Task<string[]> GetSubscriberAddresses(string topic)
         {
+            if (topic.Contains('@') && _allowPublishOnAlternateExchanges)
+            {
+                var tokens = topic.Split('@');
+                if (tokens.Last() != string.Empty)
+                {
+                    return new[] {topic};
+                }
+            }
             return new[] { $"{topic}@{_topicExchangeName}" };
         }