Change Consumer to EventingBasicConsumer

src/Hangfire.SqlServer.RabbitMq/Hangfire.SqlServer.RabbitMq.csproj

@@ -74,12 +74,14 @@
       <Link>Properties\SharedAssemblyInfo.cs</Link>
     </Compile>
     <Compile Include="RabbitMqConnectionConfiguration.cs" />
+    <Compile Include="RabbitMqExtensions.cs" />
     <Compile Include="RabbitMqFetchedJob.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="RabbitMqJobQueue.cs" />
     <Compile Include="RabbitMqJobQueueProvider.cs" />
     <Compile Include="RabbitMqMonitoringApi.cs" />
     <Compile Include="RabbitMqSqlServerStorageExtensions.cs" />
+    <Compile Include="RabbitMqTransport.cs" />
   </ItemGroup>
   <ItemGroup>
     <None Include="packages.config">

src/Hangfire.SqlServer.RabbitMq/RabbitMqConnectionConfiguration.cs

@@ -10,7 +10,11 @@ public class RabbitMqConnectionConfiguration
         public const string DefaultUser = "guest";
         public const string DefaultPassword = "guest";
         public const string DefaultVirtualHost = "/";
-        public const ushort DefaultPrefetchCount = 1;
+        public const ushort DefaultPrefetchCount = 10;
+        public const int DefaultNetworkRecoveryInterval = 10;
+        public const bool DefaultTopologyRecoveryEnabled = true;
+        public const bool DefaultAutomaticRecoveryEnabled = true;
+
 
         public RabbitMqConnectionConfiguration()
             : this(DefaultHost, DefaultPort, DefaultUser, DefaultPassword)
@@ -27,11 +31,22 @@ public RabbitMqConnectionConfiguration(string host, int port)
         {
         }
 
+        private void DefaultSets()
+        {
+            VirtualHost = DefaultVirtualHost;
+            PrefetchCount = DefaultPrefetchCount;
+
+            TopologyRecoveryEnabled = DefaultTopologyRecoveryEnabled;
+            NetworkRecoveryInterval = TimeSpan.FromSeconds(DefaultNetworkRecoveryInterval);
+            AutomaticRecoveryEnabled = DefaultAutomaticRecoveryEnabled;
+        }
+
         public RabbitMqConnectionConfiguration(Uri uri)
         {
             if (uri == null) throw new ArgumentNullException("uri");
 
             Uri = uri;
+            DefaultSets();
         }
 
         public RabbitMqConnectionConfiguration(string host, int port, string username, string password, ushort prefetchCount = DefaultPrefetchCount)
@@ -44,8 +59,7 @@ public RabbitMqConnectionConfiguration(string host, int port, string username, s
             Username = username;
             Password = password;
             Port = port;
-            VirtualHost = DefaultVirtualHost;
-            PrefetchCount = DefaultPrefetchCount;
+            DefaultSets();
         }
 
         public string Username { get; set; }
@@ -73,5 +87,11 @@ public RabbitMqConnectionConfiguration(string host, int port, string username, s
         /// establish a per-queue concurrency constraint of only one job execution at a time!
         /// </remarks>
         public ushort PrefetchCount { get; set; }
+
+        public bool TopologyRecoveryEnabled { get; set; }
+
+        public bool AutomaticRecoveryEnabled { get; set; }
+
+        public TimeSpan NetworkRecoveryInterval { get; set; }
     }
 }

src/Hangfire.SqlServer.RabbitMq/RabbitMqExtensions.cs

@@ -0,0 +1,20 @@
+using Hangfire.SqlServer;
+using Hangfire.SqlServer.RabbitMQ;
+using Hangfire.States;
+using System;
+
+namespace Hangfire
+{
+    public static class RabbitMqExtensions
+    {
+
+        public static IGlobalConfiguration<SqlServerStorage> UseRabbitMQQueues(
+            this IGlobalConfiguration<SqlServerStorage> configuration,
+            Action<RabbitMqConnectionConfiguration> configureAction,
+            params string[] queues)
+        {
+            configuration.Entry.UseRabbitMq(configureAction, queues);
+            return configuration;
+        }
+    }
+}

src/Hangfire.SqlServer.RabbitMq/RabbitMqFetchedJob.cs

@@ -2,53 +2,61 @@
 using Hangfire.Annotations;
 using Hangfire.Logging;
 using Hangfire.Storage;
+using System.Threading;
 
 namespace Hangfire.SqlServer.RabbitMQ
 {
     internal class RabbitMqFetchedJob : IFetchedJob
     {
-        private readonly Action _removeFromQueue;
-        private readonly Action _requeue;
+        private readonly Func<bool> _removeFromQueue;
+        private readonly Func<bool> _requeue;
         private bool _completed;
         private bool _disposed;
+        private readonly CancellationToken _cancellationToken;
 
         private static readonly Hangfire.Logging.ILog Logger = Hangfire.Logging.LogProvider.For<RabbitMqJobQueue>();
 
-        public RabbitMqFetchedJob(string jobId, [NotNull] Action removeFromQueue, [NotNull] Action requeue)
+        public RabbitMqFetchedJob(string jobId, [NotNull] Func<bool> removeFromQueue, [NotNull] Func<bool> requeue, CancellationToken cancellationToken)
         {
             if (removeFromQueue == null) throw new ArgumentNullException(nameof(removeFromQueue));
             if (requeue == null) throw new ArgumentNullException(nameof(requeue));
             _removeFromQueue = removeFromQueue;
             _requeue = requeue;
+            _cancellationToken = cancellationToken;
 
             JobId = jobId;
-            Logger.Debug($"Job dequeued: {JobId}");
+            Logger.Debug($"Background process '{Thread.CurrentThread.Name}': dequeued Job#{JobId}");
         }
 
         public string JobId { get; private set; }
 
         public void RemoveFromQueue()
         {
             if (_completed) throw new InvalidOperationException("Job already completed");
-            _removeFromQueue();
-            _completed = true;
-            Logger.Debug($"Job ACK'ed: {JobId}");
+            if (!_cancellationToken.IsCancellationRequested)
+            {
+                if (_removeFromQueue())
+                {
+                    _completed = true;
+                    Logger.Debug($"Background process '{Thread.CurrentThread.Name}': ack'ed: Job#{JobId}");
+                }
+            }
         }
 
         public void Requeue()
         {
             if (_completed) throw new InvalidOperationException("Job already completed");
-            _requeue();
-
-            _completed = true;
+            if (_requeue())
+            {
+                _completed = true;
+                Logger.Debug($"Background process '{Thread.CurrentThread.Name}': requeue'ed: Job#{JobId}");
+            }
         }
 
         public void Dispose()
         {
             if (!_completed && !_disposed)
-            {
                 Requeue();
-            }
 
             _disposed = true;
         }