cleanup code.

src/DotNetCore.CAP.EntityFrameworkCore/CAP.EFOptions.cs

@@ -1,6 +1,4 @@
 using System;
-using System.Collections.Generic;
-using System.Text;
 
 namespace DotNetCore.CAP
 {
@@ -29,4 +27,4 @@ namespace DotNetCore.CAP
 
         public Type DbContextType { get; internal set; }
     }
-}
+}

src/DotNetCore.CAP.EntityFrameworkCore/CAP.SqlServerCapOptionsExtension.cs

@@ -28,4 +28,4 @@ namespace DotNetCore.CAP
             services.AddSingleton(sqlServerOptions);
         }
     }
-}
+}

src/DotNetCore.CAP.EntityFrameworkCore/CAP.SqlServerOptions.cs

@@ -1,9 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-using DotNetCore.CAP.EntityFrameworkCore;
-
-namespace DotNetCore.CAP
+namespace DotNetCore.CAP
 {
     public class SqlServerOptions : EFOptions
     {
@@ -11,6 +6,5 @@ namespace DotNetCore.CAP
         /// Gets or sets the database's connection string that will be used to store database entities.
         /// </summary>
         public string ConnectionString { get; set; } //= "Server=DESKTOP-M9R8T31;Initial Catalog=Test;User Id=sa;Password=P@ssw0rd;MultipleActiveResultSets=True";
-
     }
-}
+}

src/DotNetCore.CAP.EntityFrameworkCore/CapPublisher.cs

@@ -93,4 +93,4 @@ namespace DotNetCore.CAP.EntityFrameworkCore
             PublishQueuer.PulseEvent.Set();
         }
     }
-}
+}

src/DotNetCore.CAP.EntityFrameworkCore/FetchedMessage.cs

@@ -1,7 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-using DotNetCore.CAP.Models;
+using DotNetCore.CAP.Models;
 
 namespace DotNetCore.CAP.EntityFrameworkCore
 {
@@ -11,4 +8,4 @@ namespace DotNetCore.CAP.EntityFrameworkCore
 
         public MessageType MessageType { get; set; }
     }
-}
+}

src/DotNetCore.CAP.EntityFrameworkCore/IAdditionalProcessor.Default.cs

@@ -1,11 +1,8 @@
 using System;
-using System.Collections.Generic;
 using System.Data.SqlClient;
-using System.Text;
 using System.Threading.Tasks;
 using Dapper;
 using DotNetCore.CAP.Processor;
-using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 
 namespace DotNetCore.CAP.EntityFrameworkCore
@@ -44,14 +41,14 @@ namespace DotNetCore.CAP.EntityFrameworkCore
                 var removedCount = 0;
                 do
                 {
-                    using(var connection = new SqlConnection(_options.ConnectionString))
+                    using (var connection = new SqlConnection(_options.ConnectionString))
                     {
                         removedCount = await connection.ExecuteAsync($@"
 DELETE TOP (@count)
 FROM [{_options.Schema}].[{table}] WITH (readpast)
 WHERE ExpiresAt < @now;", new { now = DateTime.Now, count = MaxBatch });
                     }
-                    
+
                     if (removedCount != 0)
                     {
                         await context.WaitAsync(_delay);
@@ -63,4 +60,4 @@ WHERE ExpiresAt < @now;", new { now = DateTime.Now, count = MaxBatch });
             await context.WaitAsync(_waitingInterval);
         }
     }
-}
+}

src/DotNetCore.CAP.EntityFrameworkCore/SqlServerFetchedMessage.cs

@@ -1,11 +1,8 @@
 using System;
-using System.Collections.Generic;
 using System.Data;
-using System.Text;
 using System.Threading;
 using Dapper;
 using DotNetCore.CAP.Models;
-using Microsoft.EntityFrameworkCore.Storage;
 
 namespace DotNetCore.CAP.EntityFrameworkCore
 {
@@ -73,4 +70,4 @@ namespace DotNetCore.CAP.EntityFrameworkCore
             }
         }
     }
-}
+}

src/DotNetCore.CAP.EntityFrameworkCore/SqlServerStorage.cs

@@ -43,7 +43,6 @@ namespace DotNetCore.CAP.EntityFrameworkCore
 
         protected virtual string CreateDbTablesScript(string schema)
         {
-
             var batchSQL =
 $@"
 IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = '{schema}')
@@ -72,7 +71,7 @@ CREATE TABLE [{schema}].[Received](
 	[Added] [datetime2](7) NOT NULL,
     [ExpiresAt] [datetime2](7) NULL,
 	[StatusName] [nvarchar](50) NOT NULL,
- CONSTRAINT [PK_{schema}.Received] PRIMARY KEY CLUSTERED 
+ CONSTRAINT [PK_{schema}.Received] PRIMARY KEY CLUSTERED
 (
 	[Id] ASC
 )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
@@ -90,7 +89,7 @@ CREATE TABLE [{schema}].[Published](
 	[Added] [datetime2](7) NOT NULL,
     [ExpiresAt] [datetime2](7) NULL,
 	[StatusName] [nvarchar](50) NOT NULL,
- CONSTRAINT [PK_{schema}.Published] PRIMARY KEY CLUSTERED 
+ CONSTRAINT [PK_{schema}.Published] PRIMARY KEY CLUSTERED
 (
 	[Id] ASC
 )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
@@ -100,4 +99,4 @@ GO";
             return batchSQL;
         }
     }
-}
+}

src/DotNetCore.CAP.Kafka/CAP.KafkaCapOptionsExtension.cs

@@ -25,4 +25,4 @@ namespace DotNetCore.CAP
             services.AddTransient<IQueueExecutor, PublishQueueExecutor>();
         }
     }
-}
+}

src/DotNetCore.CAP.Kafka/CAP.KafkaOptions.cs

@@ -32,9 +32,9 @@ namespace DotNetCore.CAP
 
         internal IEnumerable<KeyValuePair<string, object>> AsRdkafkaConfig()
         {
-            if (MainConfig.ContainsKey("bootstrap.servers")) 
+            if (MainConfig.ContainsKey("bootstrap.servers"))
                 return MainConfig.AsEnumerable();
-            
+
             if (string.IsNullOrEmpty(Servers))
             {
                 throw new ArgumentNullException(nameof(Servers));

src/DotNetCore.CAP.Kafka/CAP.Options.Extensions.cs

@@ -1,6 +1,5 @@
 using System;
 using DotNetCore.CAP;
-using DotNetCore.CAP.Kafka;
 
 namespace Microsoft.Extensions.DependencyInjection
 {

src/DotNetCore.CAP.Kafka/KafkaConsumerClient.cs

@@ -81,7 +81,6 @@ namespace DotNetCore.CAP.Kafka
             MessageReceieved?.Invoke(sender, message);
         }
 
-
         #endregion private methods
     }
 }

src/DotNetCore.CAP.Kafka/PublishQueueExecutor.cs

@@ -51,4 +51,4 @@ namespace DotNetCore.CAP.Kafka
             }
         }
     }
-}
+}

src/DotNetCore.CAP.RabbitMQ/CAP.RabbiMQOptions.cs

@@ -72,7 +72,7 @@
         public int SocketWriteTimeout { get; set; } = DefaultConnectionTimeout;
 
         /// <summary>
-        /// The port to connect on. 
+        /// The port to connect on.
         /// </summary>
         public int Port { get; set; } = -1;
     }

src/DotNetCore.CAP.RabbitMQ/CAP.RabbitMQCapOptionsExtension.cs

@@ -26,4 +26,4 @@ namespace DotNetCore.CAP
             services.AddTransient<IQueueExecutor, PublishQueueExecutor>();
         }
     }
-}
+}

src/DotNetCore.CAP.RabbitMQ/CapSubscribeAttribute.cs

@@ -6,7 +6,6 @@ namespace DotNetCore.CAP.RabbitMQ
     {
         public CapSubscribeAttribute(string name) : base(name)
         {
-
         }
     }
 }

src/DotNetCore.CAP.RabbitMQ/PublishQueueExecutor.cs

@@ -64,7 +64,6 @@ namespace DotNetCore.CAP.RabbitMQ
                         Description = ex.Message
                     }));
             }
-
         }
     }
-}
+}

src/DotNetCore.CAP/Abstractions/ModelBinding/IModelBinder.cs

@@ -14,7 +14,7 @@ namespace DotNetCore.CAP.Abstractions.ModelBinding
         /// <returns>
         /// <para>
         /// A <see cref="Task"/> which will complete when the model binding process completes.
-        /// </para> 
+        /// </para>
         /// </returns>
         Task BindModelAsync(ModelBindingContext bindingContext);
     }

src/DotNetCore.CAP/CAP.Options.cs

@@ -35,7 +35,6 @@ namespace DotNetCore.CAP
         /// </summary>
         public int PollingDelay { get; set; } = 8;
 
-
         /// <summary>
 		/// Registers an extension that will be executed when building services.
 		/// </summary>

src/DotNetCore.CAP/CAP.ServiceCollectionExtensions.cs

@@ -47,12 +47,12 @@ namespace Microsoft.Extensions.DependencyInjection
             //Processors
             services.AddTransient<PublishQueuer>();
             services.AddTransient<SubscribeQueuer>();
-            services.AddTransient<IDispatcher, DefaultDispatcher>();          
+            services.AddTransient<IDispatcher, DefaultDispatcher>();
 
             //Executors
             services.AddSingleton<IQueueExecutorFactory, QueueExecutorFactory>();
             services.AddSingleton<IQueueExecutor, SubscibeQueueExecutor>();
-            
+
             //Options
             var options = new CapOptions();
             setupAction(options);

src/DotNetCore.CAP/ICapOptionsExtension.cs

@@ -6,4 +6,4 @@ namespace DotNetCore.CAP
     {
         void AddServices(IServiceCollection services);
     }
-}
+}

src/DotNetCore.CAP/ICapPublisher.cs

@@ -1,5 +1,4 @@
-using System;
-using System.Data;
+using System.Data;
 using System.Threading.Tasks;
 
 namespace DotNetCore.CAP
@@ -21,7 +20,7 @@ namespace DotNetCore.CAP
         Task PublishAsync(string name, string content);
 
         /// <summary>
-        /// Publis a object message to specified topic.        
+        /// Publis a object message to specified topic.
         /// <para>
         ///  If you are using the EntityFramework, you need to configure the DbContextType first.
         ///  otherwise you need to use overloaded method with IDbConnection and IDbTransaction.

src/DotNetCore.CAP/IConsumerHandler.Default.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using DotNetCore.CAP.Infrastructure;

src/DotNetCore.CAP/IQueueExecutor.cs

@@ -1,7 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-using System.Threading.Tasks;
+using System.Threading.Tasks;
 
 namespace DotNetCore.CAP
 {
@@ -9,4 +6,4 @@ namespace DotNetCore.CAP
     {
         Task<OperateResult> ExecuteAsync(IStorageConnection connection, IFetchedMessage message);
     }
-}
+}

src/DotNetCore.CAP/IStorage.cs

@@ -3,14 +3,14 @@ using System.Threading.Tasks;
 
 namespace DotNetCore.CAP
 {
-	/// <summary>
-	/// Represents a persisted storage.
-	/// </summary>
-	public interface IStorage
-	{
-		/// <summary>
-		/// Initializes the storage. For example, making sure a database is created and migrations are applied.
-		/// </summary>
-		Task InitializeAsync(CancellationToken cancellationToken);
-	}
-}
+    /// <summary>
+    /// Represents a persisted storage.
+    /// </summary>
+    public interface IStorage
+    {
+        /// <summary>
+        /// Initializes the storage. For example, making sure a database is created and migrations are applied.
+        /// </summary>
+        Task InitializeAsync(CancellationToken cancellationToken);
+    }
+}

src/DotNetCore.CAP/IStorageTransaction.cs

@@ -4,16 +4,16 @@ using DotNetCore.CAP.Models;
 
 namespace DotNetCore.CAP
 {
-	public interface IStorageTransaction : IDisposable
-	{
-		void UpdateMessage(CapPublishedMessage message);
+    public interface IStorageTransaction : IDisposable
+    {
+        void UpdateMessage(CapPublishedMessage message);
 
-		void UpdateMessage(CapReceivedMessage message);
+        void UpdateMessage(CapReceivedMessage message);
 
-		void EnqueueMessage(CapPublishedMessage message);
+        void EnqueueMessage(CapPublishedMessage message);
 
-		void EnqueueMessage(CapReceivedMessage message);
+        void EnqueueMessage(CapReceivedMessage message);
 
-		Task CommitAsync();
-	}
-}
+        Task CommitAsync();
+    }
+}

src/DotNetCore.CAP/Infrastructure/Helper.cs

@@ -40,7 +40,7 @@ namespace DotNetCore.CAP.Infrastructure
         public static long ToTimestamp(DateTime value)
         {
             var elapsedTime = value - Epoch;
-            return (long) elapsedTime.TotalSeconds;
+            return (long)elapsedTime.TotalSeconds;
         }
 
         public static DateTime FromTimestamp(long value)

src/DotNetCore.CAP/Infrastructure/StatusName.cs

@@ -1,8 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace DotNetCore.CAP.Infrastructure
+namespace DotNetCore.CAP.Infrastructure
 {
     /// <summary>
     /// The message status name.
@@ -15,4 +11,4 @@ namespace DotNetCore.CAP.Infrastructure
         public const string Succeeded = nameof(Succeeded);
         public const string Failed = nameof(Failed);
     }
-}
+}

src/DotNetCore.CAP/Infrastructure/WaitHandleEx.cs

@@ -21,7 +21,7 @@ namespace DotNetCore.CAP.Infrastructure
                 var tcs = new TaskCompletionSource<bool>();
                 registeredHandle = ThreadPool.RegisterWaitForSingleObject(
                     handle,
-                    (state, timedOut) => ((TaskCompletionSource<bool>) state).TrySetResult(!timedOut),
+                    (state, timedOut) => ((TaskCompletionSource<bool>)state).TrySetResult(!timedOut),
                     tcs,
                     timeout,
                     true);

src/DotNetCore.CAP/Internal/IConsumerServiceSelector.Default.cs

@@ -44,7 +44,6 @@ namespace DotNetCore.CAP.Internal
             return executorDescriptorList;
         }
 
-
         private static IEnumerable<ConsumerExecutorDescriptor> FindConsumersFromInterfaceTypes(
             IServiceProvider provider)
         {

src/DotNetCore.CAP/Internal/MethodMatcherCache.cs

@@ -1,7 +1,7 @@
 using System;
-using System.Linq;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Linq;
 using DotNetCore.CAP.Abstractions;
 
 namespace DotNetCore.CAP.Internal
@@ -41,7 +41,7 @@ namespace DotNetCore.CAP.Internal
 
         /// <summary>
         ///  Get a dictionary of specify topic candidates.
-        ///  The Key is Group name, the value is specify topic candidates. 
+        ///  The Key is Group name, the value is specify topic candidates.
         /// </summary>
         /// <param name="topicName">message topic name</param>
         public IDictionary<string, IList<ConsumerExecutorDescriptor>> GetTopicExector(string topicName)

src/DotNetCore.CAP/Internal/ObjectMethodExecutor.cs

@@ -130,7 +130,7 @@ namespace DotNetCore.CAP.Internal
 
         private static ConsumerMethodExecutor WrapVoidAction(VoidActionExecutor executor)
         {
-            return delegate(object target, object[] parameters)
+            return delegate (object target, object[] parameters)
             {
                 executor(target, parameters);
                 return null;
@@ -192,7 +192,7 @@ namespace DotNetCore.CAP.Internal
         /// </summary>
         private static async Task<object> CastToObject<T>(Task<T> task)
         {
-            return (object) await task;
+            return (object)await task;
         }
 
         private static Type GetTaskInnerTypeOrNull(Type type)
@@ -279,7 +279,7 @@ namespace DotNetCore.CAP.Internal
 
         private static Task<object> Convert<T>(object taskAsObject)
         {
-            var task = (Task<T>) taskAsObject;
+            var task = (Task<T>)taskAsObject;
             return CastToObject<T>(task);
         }
 

src/DotNetCore.CAP/Internal/SubscriberNotFoundException.cs

@@ -1,13 +1,19 @@
 using System;
+
 namespace DotNetCore.CAP.Internal
 {
     public class SubscriberNotFoundException : Exception
     {
-        public SubscriberNotFoundException() { }
+        public SubscriberNotFoundException()
+        {
+        }
 
-        public SubscriberNotFoundException(string message) : base(message) { }
+        public SubscriberNotFoundException(string message) : base(message)
+        {
+        }
 
         public SubscriberNotFoundException(string message, Exception inner) :
-            base(message, inner) { }
+            base(message, inner)
+        { }
     }
-}
+}

src/DotNetCore.CAP/LoggerExtensions.cs

@@ -1,7 +1,4 @@
 using System;
-using System.Collections.Generic;
-using System.Linq;
-using DotNetCore.CAP.Processor;
 using Microsoft.Extensions.Logging;
 
 namespace DotNetCore.CAP
@@ -66,7 +63,6 @@ namespace DotNetCore.CAP
                 5,
                 "Received message topic method '{topicName}' failed to execute.");
 
-
             _jobRetrying = LoggerMessage.Define<int>(
                 LogLevel.Debug,
                 3,
@@ -104,19 +100,16 @@ namespace DotNetCore.CAP
             _jobFailedWillRetry(logger, ex);
         }
 
-
         public static void JobRetrying(this ILogger logger, int retries)
         {
             _jobRetrying(logger, retries, null);
         }
 
-
         public static void JobExecuted(this ILogger logger, double seconds)
         {
             _jobExecuted(logger, seconds, null);
         }
 
-
         public static void ConsumerMethodExecutingFailed(this ILogger logger, string methodName, Exception ex)
         {
             _executingConsumerMethod(logger, methodName, ex);

src/DotNetCore.CAP/Models/CapPublishedMessage.cs

@@ -3,7 +3,7 @@ using DotNetCore.CAP.Infrastructure;
 
 namespace DotNetCore.CAP.Models
 {
-    public class CapPublishedMessage 
+    public class CapPublishedMessage
     {
         /// <summary>
         /// Initializes a new instance of <see cref="CapPublishedMessage"/>.
@@ -22,7 +22,7 @@ namespace DotNetCore.CAP.Models
             Content = message.Content;
         }
 
-        public int Id { get; set; } 
+        public int Id { get; set; }
 
         public string Name { get; set; }
 

src/DotNetCore.CAP/Models/CapQueue.cs

@@ -9,4 +9,4 @@
         /// </summary>
         public MessageType MessageType { get; set; }
     }
-}
+}

src/DotNetCore.CAP/Models/MessageType.cs

@@ -5,4 +5,4 @@
         Publish,
         Subscribe
     }
-}
+}

src/DotNetCore.CAP/OperateResult.cs

@@ -32,7 +32,7 @@ namespace DotNetCore.CAP
         /// Returns an <see cref="OperateResult"/> indicating a successful identity operation.
         /// </summary>
         /// <returns>An <see cref="OperateResult"/> indicating a successful operation.</returns>
-        public static OperateResult Success { get; } = new OperateResult {Succeeded = true};
+        public static OperateResult Success { get; } = new OperateResult { Succeeded = true };
 
         /// <summary>
         /// Creates an <see cref="OperateResult"/> indicating a failed operation, with a list of <paramref name="errors"/> if applicable.
@@ -41,7 +41,7 @@ namespace DotNetCore.CAP
         /// <returns>An <see cref="OperateResult"/> indicating a failed operation, with a list of <paramref name="errors"/> if applicable.</returns>
         public static OperateResult Failed(params OperateError[] errors)
         {
-            var result = new OperateResult {Succeeded = false};
+            var result = new OperateResult { Succeeded = false };
             if (errors != null)
             {
                 result._errors.AddRange(errors);

src/DotNetCore.CAP/Processor/IAdditionalProcessor.cs

@@ -1,11 +1,6 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace DotNetCore.CAP.Processor
+namespace DotNetCore.CAP.Processor
 {
     public interface IAdditionalProcessor : IProcessor
     {
-
     }
-}
+}

src/DotNetCore.CAP/Processor/IDispatcher.Default.cs

@@ -93,4 +93,4 @@ namespace DotNetCore.CAP.Processor
             return fetched != null;
         }
     }
-}
+}

src/DotNetCore.CAP/Processor/IDispatcher.cs

@@ -1,11 +1,7 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace DotNetCore.CAP.Processor
+namespace DotNetCore.CAP.Processor
 {
     public interface IDispatcher : IProcessor
     {
         bool Waiting { get; }
     }
-}
+}

src/DotNetCore.CAP/Processor/IProcessor.PublishQueuer.cs

@@ -2,67 +2,67 @@
 using System.Threading;
 using System.Threading.Tasks;
 using DotNetCore.CAP.Infrastructure;
-using DotNetCore.CAP.Processor.States;
 using DotNetCore.CAP.Models;
+using DotNetCore.CAP.Processor.States;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
 namespace DotNetCore.CAP.Processor
 {
-	public class PublishQueuer : IProcessor
+    public class PublishQueuer : IProcessor
     {
-		private ILogger _logger;
-		private CapOptions _options;
-		private IStateChanger _stateChanger;
-		private IServiceProvider _provider;
-		private TimeSpan _pollingDelay;
+        private ILogger _logger;
+        private CapOptions _options;
+        private IStateChanger _stateChanger;
+        private IServiceProvider _provider;
+        private TimeSpan _pollingDelay;
 
         public static readonly AutoResetEvent PulseEvent = new AutoResetEvent(true);
 
         public PublishQueuer(
-			ILogger<PublishQueuer> logger,
-			IOptions<CapOptions> options,
-			IStateChanger stateChanger,
-			IServiceProvider provider)
-		{
-			_logger = logger;
-			_options = options.Value;
-			_stateChanger = stateChanger;
-			_provider = provider;
+            ILogger<PublishQueuer> logger,
+            IOptions<CapOptions> options,
+            IStateChanger stateChanger,
+            IServiceProvider provider)
+        {
+            _logger = logger;
+            _options = options.Value;
+            _stateChanger = stateChanger;
+            _provider = provider;
 
-			_pollingDelay = TimeSpan.FromSeconds(_options.PollingDelay);
-		}
+            _pollingDelay = TimeSpan.FromSeconds(_options.PollingDelay);
+        }
 
-		public async Task ProcessAsync(ProcessingContext context)
-		{
-			using (var scope = _provider.CreateScope())
-			{
+        public async Task ProcessAsync(ProcessingContext context)
+        {
+            using (var scope = _provider.CreateScope())
+            {
                 CapPublishedMessage sentMessage;
-				var provider = scope.ServiceProvider;
-				var connection = provider.GetRequiredService<IStorageConnection>();
+                var provider = scope.ServiceProvider;
+                var connection = provider.GetRequiredService<IStorageConnection>();
 
-				while (
-					!context.IsStopping &&
-					(sentMessage = await connection.GetNextPublishedMessageToBeEnqueuedAsync()) != null)
+                while (
+                    !context.IsStopping &&
+                    (sentMessage = await connection.GetNextPublishedMessageToBeEnqueuedAsync()) != null)
 
                 {
                     var state = new EnqueuedState();
 
-					using (var transaction = connection.CreateTransaction())
-					{
-						_stateChanger.ChangeState(sentMessage, state, transaction);
-						await transaction.CommitAsync();
-					}
-				} 
-			}
+                    using (var transaction = connection.CreateTransaction())
+                    {
+                        _stateChanger.ChangeState(sentMessage, state, transaction);
+                        await transaction.CommitAsync();
+                    }
+                }
+            }
 
             context.ThrowIfStopping();
-            
+
             DefaultDispatcher.PulseEvent.Set();
 
             await WaitHandleEx.WaitAnyAsync(PulseEvent,
                 context.CancellationToken.WaitHandle, _pollingDelay);
-		}
-	}
-}
+        }
+    }
+}

src/DotNetCore.CAP/Processor/IProcessor.SubscribeQueuer.cs

@@ -2,8 +2,8 @@
 using System.Threading;
 using System.Threading.Tasks;
 using DotNetCore.CAP.Infrastructure;
-using DotNetCore.CAP.Processor.States;
 using DotNetCore.CAP.Models;
+using DotNetCore.CAP.Processor.States;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
@@ -65,4 +65,4 @@ namespace DotNetCore.CAP.Processor
                 context.CancellationToken.WaitHandle, _pollingDelay);
         }
     }
-}
+}

src/DotNetCore.CAP/Processor/RetryBehavior.cs

@@ -18,7 +18,7 @@ namespace DotNetCore.CAP.Processor
         {
             DefaultRetryCount = 25;
             DefaultRetryInThunk = retries =>
-                (int) Math.Round(Math.Pow(retries - 1, 4) + 15 + (_random.Next(30) * (retries)));
+                (int)Math.Round(Math.Pow(retries - 1, 4) + 15 + (_random.Next(30) * (retries)));
 
             DefaultRetry = new RetryBehavior(true);
             NoRetry = new RetryBehavior(false);

src/DotNetCore.CAP/QueueExecutorFactory.cs

@@ -18,8 +18,8 @@ namespace DotNetCore.CAP
         public IQueueExecutor GetInstance(MessageType messageType)
         {
             var _queueExectors = _serviceProvider.GetServices<IQueueExecutor>();
-            
-            if (messageType== MessageType.Publish)
+
+            if (messageType == MessageType.Publish)
             {
                 return _queueExectors.FirstOrDefault(x => typeof(BasePublishQueueExecutor).IsAssignableFrom(x.GetType()));
             }
@@ -29,4 +29,4 @@ namespace DotNetCore.CAP
             }
         }
     }
-}
+}