Move poison messages to the error queue

# Conflicts:
#	src/NServiceBus.RabbitMQ.Tests/RabbitMqContext.cs

Author: bording

src/NServiceBus.RabbitMQ.Tests/ClusteringTests/ClusteredTestContext.cs

@@ -22,6 +22,7 @@
     public abstract class ClusteredTestContext
     {
         protected const string QueueName = "testreceiver";
+        protected const string ErrorQueue = "error";
         const string ErlangProcessName = "erl";
         protected static Logger Logger = LogManager.GetCurrentClassLogger();
 
@@ -210,7 +211,7 @@ protected void SetupQueueAndSenderAndListener(string connectionString)
         void SetupQueueListener(string queueName)
         {
             receivedMessages = new BlockingCollection<TransportMessage>();
-            dequeueStrategy = new RabbitMqDequeueStrategy(connectionManager, null, new ReceiveOptions(s => SecondaryReceiveSettings.Disabled(), new MessageConverter(),1,1000,true,"Cluster test"));
+            dequeueStrategy = new RabbitMqDequeueStrategy(connectionManager, null, new ReceiveOptions(s => SecondaryReceiveSettings.Disabled(), new MessageConverter(),1,1000,true,"Cluster test"), ErrorQueue);
             dequeueStrategy.Init(Address.Parse(queueName), new TransactionSettings(true, TimeSpan.FromSeconds(30), IsolationLevel.ReadCommitted, 5, false, false), m =>
                 {
                     receivedMessages.Add(m);

src/NServiceBus.RabbitMQ.Tests/RabbitMqContext.cs

@@ -79,10 +79,11 @@ public void SetUp()
             sender = new RabbitMqMessageSender(routingTopology, channelProvider, new IncomingContext(null, null));
 
             dequeueStrategy = new RabbitMqDequeueStrategy(connectionManager, new RepeatedFailuresOverTimeCircuitBreaker("UnitTest",TimeSpan.FromMinutes(2),e=>{}),
-                new ReceiveOptions(s => SecondaryReceiveSettings.Enabled(CallbackQueue, 1), new MessageConverter(),1,1000,false,"Unit test"));
+                new ReceiveOptions(s => SecondaryReceiveSettings.Enabled(CallbackQueue, 1), new MessageConverter(),1,1000,false,"Unit test"), ErrorQueue);
 
 
             MakeSureQueueAndExchangeExists(ReceiverQueue);
+            MakeSureQueueAndExchangeExists(ErrorQueue);
 
 
             MessagePublisher = new RabbitMqMessagePublisher
@@ -146,6 +147,7 @@ protected TransportMessage WaitForMessage()
         protected string CallbackQueue = "testreceiver." + RuntimeEnvironment.MachineName;
 
         protected const string ReceiverQueue = "testreceiver";
+        protected const string ErrorQueue = "error";
         protected RabbitMqMessagePublisher MessagePublisher;
         protected RabbitMqConnectionManager connectionManager;
         protected RabbitMqDequeueStrategy dequeueStrategy;
@@ -229,4 +231,4 @@ public ConfirmsAwareChannel GetNewPublishChannel()
             throw new NotImplementedException();
         }
     }
-}
+}

src/NServiceBus.RabbitMQ/Config/RabbitMqTransportFeature.cs

@@ -2,15 +2,16 @@
 {
     using System;
     using System.Configuration;
-    using NServiceBus.CircuitBreakers;
-    using NServiceBus.Pipeline;
-    using NServiceBus.Transports.RabbitMQ.Connection;
+    using CircuitBreakers;
+    using Config;
+    using Pipeline;
     using RabbitMQ.Client.Events;
     using Settings;
     using Support;
     using Transports;
     using Transports.RabbitMQ;
     using Transports.RabbitMQ.Config;
+    using Transports.RabbitMQ.Connection;
     using Transports.RabbitMQ.Routing;
 
     class RabbitMqTransportFeature : ConfigureTransport
@@ -41,6 +42,7 @@ protected override void Configure(FeatureConfigurationContext context, string co
             var queueName = GetLocalAddress(context.Settings);
             var callbackQueue = string.Format("{0}.{1}", queueName, RuntimeEnvironment.MachineName);
             var connectionConfiguration = new ConnectionStringParser(context.Settings).Parse(connectionString);
+            var errorQueue = context.Settings.GetConfigSection<MessageForwardingInCaseOfFaultConfig>()?.ErrorQueue;
 
             MessageConverter messageConverter;
 
@@ -83,7 +85,8 @@ protected override void Configure(FeatureConfigurationContext context, string co
             context.Container.ConfigureComponent(builder => new RabbitMqDequeueStrategy(
                 builder.Build<IManageRabbitMqConnections>(),
                 SetupCircuitBreaker(builder.Build<CriticalError>()),
-                receiveOptions), DependencyLifecycle.InstancePerCall);
+                receiveOptions,
+                errorQueue), DependencyLifecycle.InstancePerCall);
 
 
             context.Container.ConfigureComponent<OpenPublishChannelBehavior>(DependencyLifecycle.InstancePerCall);

src/NServiceBus.RabbitMQ/RabbitMqDequeueStrategy.cs

@@ -13,11 +13,12 @@
 
     class RabbitMqDequeueStrategy : IDequeueMessages, IDisposable
     {
-        public RabbitMqDequeueStrategy(IManageRabbitMqConnections connectionManager, RepeatedFailuresOverTimeCircuitBreaker circuitBreaker, ReceiveOptions receiveOptions)
+        public RabbitMqDequeueStrategy(IManageRabbitMqConnections connectionManager, RepeatedFailuresOverTimeCircuitBreaker circuitBreaker, ReceiveOptions receiveOptions, string errorQueue)
         {
             this.connectionManager = connectionManager;
             this.circuitBreaker = circuitBreaker;
             this.receiveOptions = receiveOptions;
+            this.errorQueue = errorQueue;
         }
 
         public void Init(Address address, TransactionSettings transactionSettings, Func<TransportMessage, bool> tryProcessMessage, Action<TransportMessage, Exception> endProcessMessage)
@@ -187,10 +188,8 @@ void ConsumeMessages(object state)
                             }
                             catch (Exception ex)
                             {
-                                Logger.Error("Poison message detected, deliveryTag: " + message.DeliveryTag, ex);
-
-                                //just ack the poison message to avoid getting stuck
-                                messageProcessedOk = true;
+                                Logger.Error($"Poison message detected. Moving message to queue '{errorQueue}'...", ex);
+                                messageProcessedOk = MovePoisonMessage(message, errorQueue);
                             }
 
                             if (transportMessage != null)
@@ -276,7 +275,26 @@ void Purge()
             }
         }
 
+        bool MovePoisonMessage(BasicDeliverEventArgs message, string queue)
+        {
+            var success = false;
+            var connection = connectionManager.GetPublishConnection();
+
+            using (var channel = connection.CreateModel())
+            {
+                try
+                {
+                    channel.BasicPublish("", queue, false, message.BasicProperties, message.Body);
+                    success = true;
+                }
+                catch (Exception ex)
+                {
+                    Logger.Error($"Failed to move poison message to queue '{queue}'. Returning message to original queue...", ex);
+                }
+            }
 
+            return success;
+        }
 
         static ILog Logger = LogManager.GetLogger(typeof(RabbitMqDequeueStrategy));
 
@@ -293,7 +311,7 @@ void Purge()
         readonly ReceiveOptions receiveOptions;
         ushort actualPrefetchCount;
         bool isStopping;
-
+        readonly string errorQueue;
 
         class ConsumeParams
         {