Allow infinite timeout with publish confirms (#1049)

Author: Pliner

Source/EasyNetQ/ConnectionConfigurationExtensions.cs

@@ -0,0 +1,14 @@
+using System;
+using System.Threading;
+
+namespace EasyNetQ
+{
+    // TODO Should migrate Timeout from ushort to TimeSpan
+    internal static class ConnectionConfigurationExtensions
+    {
+        public static TimeSpan GetTimeout(this ConnectionConfiguration configuration)
+        {
+            return configuration.Timeout == 0 ? Timeout.InfiniteTimeSpan : TimeSpan.FromSeconds(configuration.Timeout);
+        }
+    }
+}

Source/EasyNetQ/Producer/PersistentChannel.cs

@@ -42,9 +42,7 @@ public void InvokeChannelAction(Action<IModel> channelAction)
         {
             Preconditions.CheckNotNull(channelAction, "channelAction");
 
-            var timeout = configuration.Timeout.Equals(0)
-                ? TimeBudget.Infinite()
-                : TimeBudget.Start(TimeSpan.FromSeconds(configuration.Timeout));
+            var timeout = TimeBudget.Start(configuration.GetTimeout());
 
             var retryTimeoutMs = MinRetryTimeoutMs;
             while (!timeout.IsExpired())

Source/EasyNetQ/RabbitAdvancedBus.cs

@@ -228,9 +228,14 @@ public void Publish(
             var rawMessage = produceConsumeInterceptor.OnProduce(new RawMessage(messageProperties, body));
             if (connectionConfiguration.PublisherConfirms)
             {
-                var timeout = TimeBudget.Start(TimeSpan.FromSeconds(connectionConfiguration.Timeout));
-                while (!timeout.IsExpired())
+                var timeout = TimeBudget.Start(connectionConfiguration.GetTimeout());
+                while (true)
                 {
+                    if (timeout.IsExpired())
+                    {
+                        throw new TimeoutException($"Publish timed out after {connectionConfiguration.Timeout} seconds");
+                    }
+
                     var confirmsWaiter = clientCommandDispatcher.Invoke(model =>
                     {
                         var properties = model.CreateBasicProperties();
@@ -338,9 +343,14 @@ public virtual async Task PublishAsync(
             var rawMessage = produceConsumeInterceptor.OnProduce(new RawMessage(messageProperties, body));
             if (connectionConfiguration.PublisherConfirms)
             {
-                var timeout = TimeBudget.Start(TimeSpan.FromSeconds(connectionConfiguration.Timeout));
-                while (!timeout.IsExpired())
+                var timeout = TimeBudget.Start(connectionConfiguration.GetTimeout());
+                while (true)
                 {
+                    if (timeout.IsExpired())
+                    {
+                        throw new TimeoutException($"Publish timed out after {connectionConfiguration.Timeout} seconds");
+                    }
+
                     var confirmsWaiter = await clientCommandDispatcher.InvokeAsync(model =>
                     {
                         var properties = model.CreateBasicProperties();