Release 2.2.2 (#121)

* fixed message enqueue exception in v2.2

* add log to samples

* Improved log output.  #114

* add default timeout configuration for kafka client.

* fixed retry processor bugs.

* Fixed kafka producer exception log without logging when publish message.

* update version num to 2.2.2

* rename configuration options FailedCallback to FailedThresholdCallback

* rename files name.

* remove unused files.

* modify the error comments.

* update samples.

* add logs.

build/version.props

@@ -2,7 +2,7 @@
 	<PropertyGroup>
 		<VersionMajor>2</VersionMajor>
 		<VersionMinor>2</VersionMinor>
-		<VersionPatch>1</VersionPatch>
+		<VersionPatch>2</VersionPatch>
 		<VersionQuality></VersionQuality>
 		<VersionPrefix>$(VersionMajor).$(VersionMinor).$(VersionPatch)</VersionPrefix>
 	</PropertyGroup>

samples/Sample.Kafka.MySql/Controllers/ValuesController.cs

@@ -37,7 +37,7 @@ namespace Sample.Kafka.MySql.Controllers
         [CapSubscribe("xxx.xxx.test2")]
         public void Test2(int value)
         {
-            Console.WriteLine(value);
+            Console.WriteLine("Subscriber output message: " + value);
         }
     }
 }
\ No newline at end of file

samples/Sample.Kafka.MySql/Program.cs

 using Microsoft.AspNetCore;
 using Microsoft.AspNetCore.Hosting;
+using NLog.Web;
 
 namespace Sample.Kafka.MySql
 {
@@ -14,7 +15,11 @@ namespace Sample.Kafka.MySql
         public static IWebHost BuildWebHost(string[] args) =>
             WebHost.CreateDefaultBuilder(args)
                 .UseStartup<Startup>()
+                .ConfigureLogging((hostingContext, builder) =>
+                {
+                    hostingContext.HostingEnvironment.ConfigureNLog("nlog.config");
+                })
+                .UseNLog()
                 .Build();
-
     }
 }
\ No newline at end of file

samples/Sample.Kafka.MySql/Sample.Kafka.MySql.csproj

@@ -10,6 +10,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.AspNetCore.All" Version="2.0.6" />
     <PackageReference Include="MySqlConnector" Version="0.38.0" />
+    <PackageReference Include="NLog.Web.AspNetCore" Version="4.5.2" />
     <PackageReference Include="zipkin4net" Version="1.2.0" />
   </ItemGroup>
   <ItemGroup>
@@ -20,5 +21,10 @@
     <ProjectReference Include="..\..\src\DotNetCore.CAP.MySql\DotNetCore.CAP.MySql.csproj" />
     <ProjectReference Include="..\..\src\DotNetCore.CAP\DotNetCore.CAP.csproj" />
   </ItemGroup>
+  <ItemGroup>
+    <Content Update="nlog.config">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </Content>
+  </ItemGroup>
 
 </Project>

samples/Sample.Kafka.MySql/nlog.config

+<?xml version="1.0" encoding="utf-8"?>
+
+<nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd"
+      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+      autoReload="true"
+      internalLogLevel="Warn"
+      internalLogFile="logs/internal-nlog.txt">
+
+  <variable name="myLogLayout"
+            value="---------------------------------------------------------------------------${newline}Date：${longdate}    Level：${uppercase:${level}}    User：${aspnet-user-identity}(${aspnet-request-ip})    Logger：${logger}  URL：${aspnet-request-method} ${aspnet-request-url:IncludePort=true:IncludeQueryString=true}  Action：${aspnet-mvc-action} ${newline}Message：${message} ${newline}${onexception:Exception：${exception:format=toString}}" />
+  <extensions>
+    <add assembly="NLog.Web.AspNetCore" />
+  </extensions>
+
+  <!-- define various log targets -->
+  <targets>
+    <!-- write logs to file -->
+    <target name="allfile" xsi:type="File" fileName="logs/cap-all-${shortdate}.log"
+            layout="${myLogLayout}" />
+  </targets>
+
+  <rules>
+    <!--All logs, including from Microsoft-->
+    <logger name="*" minlevel="Debug" writeTo="allfile" /> 
+  </rules>
+</nlog>
\ No newline at end of file

samples/Sample.RabbitMQ.MySql/AppDbContext.cs

-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading.Tasks;
-using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore;
 
 namespace Sample.RabbitMQ.MySql
 {
@@ -10,8 +6,7 @@ namespace Sample.RabbitMQ.MySql
     {
         protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
         {
-            optionsBuilder.UseMySql("Server=localhost;Database=Sample.RabbitMQ.MySql;UserId=root;Password=123123;Allow User Variables=True");
-            //optionsBuilder.UseMySql("Server=192.168.2.206;Database=Sample.RabbitMQ.MySql;UserId=root;Password=123123;");
+            optionsBuilder.UseMySql("Server=192.168.10.110;Database=testcap;UserId=root;Password=123123;");
         }
     }
 }

samples/Sample.RabbitMQ.MySql/Controllers/ValuesController.cs

@@ -21,7 +21,7 @@ namespace Sample.RabbitMQ.MySql.Controllers
         public IActionResult PublishMessage()
         {
             _capBus.Publish("sample.rabbitmq.mysql", DateTime.Now);
-            
+
             return Ok();
         }
 
@@ -50,7 +50,7 @@ namespace Sample.RabbitMQ.MySql.Controllers
         [CapSubscribe("sample.rabbitmq.mysql")]
         public void ReceiveMessage(DateTime time)
         {
-            Console.WriteLine("[sample.rabbitmq.mysql] message received: "+ DateTime.Now.ToString() +" , sent time: " + time.ToString());
+            Console.WriteLine("[sample.rabbitmq.mysql] message received: " + DateTime.Now + ",sent time: " + time);
         }
     }
 }

samples/Sample.RabbitMQ.MySql/Program.cs

-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Linq;
-using System.Threading.Tasks;
-using Microsoft.AspNetCore;
-using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore;
 using Microsoft.AspNetCore.Hosting;
-using Microsoft.Extensions.Configuration;
+using NLog.Web;
 
 namespace Sample.RabbitMQ.MySql
 {
@@ -20,6 +14,11 @@ namespace Sample.RabbitMQ.MySql
         public static IWebHost BuildWebHost(string[] args) =>
             WebHost.CreateDefaultBuilder(args)
                 .UseStartup<Startup>()
+                .ConfigureLogging((hostingContext, builder) =>
+                {
+                    hostingContext.HostingEnvironment.ConfigureNLog("nlog.config");
+                })
+                .UseNLog()
                 .Build();
     }
 }

samples/Sample.RabbitMQ.MySql/Sample.RabbitMQ.MySql.csproj

@@ -12,6 +12,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.AspNetCore.All" Version="2.0.6" />
     <PackageReference Include="Pomelo.EntityFrameworkCore.MySql" Version="2.0.1" />
+    <PackageReference Include="NLog.Web.AspNetCore" Version="4.5.2" />
   </ItemGroup>
   <ItemGroup>
     <DotNetCliToolReference Include="Microsoft.EntityFrameworkCore.Tools.DotNet" Version="2.0.0" />
@@ -21,5 +22,10 @@
     <ProjectReference Include="..\..\src\DotNetCore.CAP.RabbitMQ\DotNetCore.CAP.RabbitMQ.csproj" />
     <ProjectReference Include="..\..\src\DotNetCore.CAP\DotNetCore.CAP.csproj" />
   </ItemGroup>
+  <ItemGroup>
+    <Content Update="nlog.config">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </Content>
+  </ItemGroup>
 
 </Project>

samples/Sample.RabbitMQ.MySql/nlog.config

+<?xml version="1.0" encoding="utf-8"?>
+
+<nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd"
+      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+      autoReload="true"
+      internalLogLevel="Warn"
+      internalLogFile="logs/internal-nlog.txt">
+
+  <variable name="myLogLayout"
+            value="---------------------------------------------------------------------------${newline}Date：${longdate}    Level：${uppercase:${level}}    User：${aspnet-user-identity}(${aspnet-request-ip})    Logger：${logger}  URL：${aspnet-request-method} ${aspnet-request-url:IncludePort=true:IncludeQueryString=true}  Action：${aspnet-mvc-action} ${newline}Message：${message} ${newline}${onexception:Exception：${exception:format=toString}}" />
+  <extensions>
+    <add assembly="NLog.Web.AspNetCore" />
+  </extensions>
+
+  <!-- define various log targets -->
+  <targets>
+    <!-- write logs to file -->
+    <target name="allfile" xsi:type="File" fileName="logs/cap-all-${shortdate}.log"
+            layout="${myLogLayout}" />
+  </targets>
+
+  <rules>
+    <!--All logs, including from Microsoft-->
+    <logger name="*" minlevel="Debug" writeTo="allfile" />
+  </rules>
+</nlog>
\ No newline at end of file

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

@@ -57,6 +57,8 @@ namespace DotNetCore.CAP
                 MainConfig["socket.blocking.max.ms"] = "10";
                 MainConfig["enable.auto.commit"] = "false";
                 MainConfig["log.connection.close"] = "false";
+                MainConfig["request.timeout.ms"] = "3000";
+                MainConfig["message.timeout.ms"] = "5000";
 
                 _kafkaConfig = MainConfig.AsEnumerable();
             }

src/DotNetCore.CAP.Kafka/IPublishMessageSender.Kafka.cs

@@ -36,11 +36,7 @@ namespace DotNetCore.CAP.Kafka
 
                 if (message.Error.HasError)
                 {
-                    return OperateResult.Failed(new OperateError
-                    {
-                        Code = message.Error.Code.ToString(),
-                        Description = message.Error.Reason
-                    });
+                    throw new PublisherSentFailedException(message.Error.ToString());
                 }
 
                 _logger.LogDebug($"kafka topic message [{keyName}] has been published.");
@@ -61,6 +57,6 @@ namespace DotNetCore.CAP.Kafka
                     producer.Dispose();
                 }
             }
-        } 
+        }
     }
 }
\ No newline at end of file

src/DotNetCore.CAP/Abstractions/CapPublisherBase.cs

@@ -132,7 +132,7 @@ namespace DotNetCore.CAP.Abstractions
             {
                 throw new InvalidOperationException(
                     "If you are using the EntityFramework, you need to configure the DbContextType first." +
-                    " otherwise you need to use overloaded method with IDbConnection and IDbTransaction.");
+                    " otherwise you need to use overloaded method with IDbTransaction.");
             }
         }
 

src/DotNetCore.CAP/CAP.Options.cs

@@ -58,9 +58,9 @@ namespace DotNetCore.CAP
         public int FailedRetryInterval { get; set; }
 
         /// <summary>
-        /// We’ll invoke this call-back with message type,name,content when requeue failed message.
+        /// We’ll invoke this call-back with message type,name,content when retry failed (send or executed) messages equals <see cref="FailedRetryCount"/> times.
         /// </summary>
-        public Action<MessageType, string, string> FailedCallback { get; set; }
+        public Action<MessageType, string, string> FailedThresholdCallback { get; set; }
 
         /// <summary>
         /// The number of message retries, the retry will stop when the threshold is reached.

src/DotNetCore.CAP/ICapPublisher.cs

@@ -15,7 +15,7 @@ namespace DotNetCore.CAP
         /// (EntityFramework) Asynchronous publish a object message.
         /// <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.
+        /// otherwise you need to use overloaded method with  IDbTransaction.
         /// </para>
         /// </summary>
         /// <typeparam name="T">The type of content object.</typeparam>
@@ -28,7 +28,7 @@ namespace DotNetCore.CAP
         /// (EntityFramework) Publish a object message.
         /// <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.
+        /// otherwise you need to use overloaded method with IDbTransaction.
         /// </para>
         /// </summary>
         /// <typeparam name="T">The type of content object.</typeparam>

src/DotNetCore.CAP/ICapSubscribe.cs

@@ -4,7 +4,7 @@
 namespace DotNetCore.CAP
 {
     /// <summary>
-    /// An empty interface, which is used to mark the current class have a CAP methods.
+    /// An empty interface, which is used to mark the current class have a CAP subscriber methods.
     /// </summary>
     public interface ICapSubscribe
     {

src/DotNetCore.CAP/IConsumerHandler.Default.cs

@@ -178,6 +178,8 @@ namespace DotNetCore.CAP
 
         private (Guid, string) TracingBefore(string topic, string values)
         {
+            _logger.LogDebug("CAP received topic message:" + topic);
+
             Guid operationId = Guid.NewGuid();
 
             var eventData = new BrokerConsumeEventData(

src/DotNetCore.CAP/IFetchedMessage.cs

-// Copyright (c) .NET Core Community. All rights reserved.
-// Licensed under the MIT License. See License.txt in the project root for license information.
-
-using System;
-using DotNetCore.CAP.Models;
-
-namespace DotNetCore.CAP
-{
-    public interface IFetchedMessage : IDisposable
-    {
-        int MessageId { get; }
-
-        MessageType MessageType { get; }
-
-        void RemoveFromQueue();
-
-        void Requeue();
-    }
-}
\ No newline at end of file

src/DotNetCore.CAP/IPublishMessageSender.Base.cs

@@ -67,15 +67,13 @@ namespace DotNetCore.CAP
             }
             else
             {
-                TracingError(operationId, message.Name, sendValues, result.Exception, startTime, stopwatch.Elapsed);
-
-                _logger.MessagePublishException(message.Id, result.Exception);
+                TracingError(operationId, message, result, startTime, stopwatch.Elapsed);
 
                 await SetFailedState(message, result.Exception, out bool stillRetry);
 
                 if (stillRetry)
                 {
-                    _logger.SenderRetrying(3);
+                    _logger.SenderRetrying(message.Id, message.Retries);
 
                     await SendAsync(message);
                 }
@@ -109,20 +107,11 @@ namespace DotNetCore.CAP
         private Task SetFailedState(CapPublishedMessage message, Exception ex, out bool stillRetry)
         {
             IState newState = new FailedState();
-
-            if (ex is PublisherSentFailedException)
-            {
-                stillRetry = false;
-                message.Retries = _options.FailedRetryCount; // not retry if PublisherSentFailedException
-            }
-            else
+            stillRetry = UpdateMessageForRetryAsync(message);
+            if (stillRetry)
             {
-                stillRetry = UpdateMessageForRetryAsync(message);
-                if (stillRetry)
-                {
-                    _logger.ConsumerExecutionFailedWillRetry(ex);
-                    return Task.CompletedTask;
-                }
+                _logger.ConsumerExecutionFailedWillRetry(ex);
+                return Task.CompletedTask;
             }
 
             AddErrorReasonToContent(message, ex);
@@ -166,14 +155,18 @@ namespace DotNetCore.CAP
             _logger.MessageHasBeenSent(du.TotalSeconds);
         }
 
-        private void TracingError(Guid operationId, string topic, string values, Exception ex, DateTimeOffset startTime, TimeSpan du)
+        private void TracingError(Guid operationId, CapPublishedMessage message, OperateResult result, DateTimeOffset startTime, TimeSpan du)
         {
+            var ex = new PublisherSentFailedException(result.ToString(), result.Exception);
+
+            _logger.MessagePublishException(message.Id, result.ToString(), ex);
+
             var eventData = new BrokerPublishErrorEventData(
                 operationId,
                 "",
                 ServersAddress,
-                topic,
-                values,
+                message.Name,
+                message.Content,
                 ex,
                 startTime,
                 du);

src/DotNetCore.CAP/Infrastructure/Helper.cs

@@ -86,7 +86,6 @@ namespace DotNetCore.CAP.Infrastructure
             return !CanConvertFromString(type);
         }
 
-
         public static string AddExceptionProperty(string json, Exception exception)
         {
             var jObject = ToJObject(exception);

src/DotNetCore.CAP/Internal/PublisherSentFailedException.cs

@@ -7,6 +7,10 @@ namespace DotNetCore.CAP.Internal
 {
     public class PublisherSentFailedException : Exception
     {
+        public PublisherSentFailedException(string message) : base(message)
+        {
+        }
+
         public PublisherSentFailedException(string message, Exception ex) : base(message, ex)
         {
         }

src/DotNetCore.CAP/LoggerExtensions.cs

@@ -17,10 +17,10 @@ namespace DotNetCore.CAP
         private static readonly Action<ILogger, string, string, string, Exception> _modelBinderFormattingException;
         private static readonly Action<ILogger, Exception> _consumerFailedWillRetry;
         private static readonly Action<ILogger, double, Exception> _consumerExecuted;
-        private static readonly Action<ILogger, int, Exception> _senderRetrying;
+        private static readonly Action<ILogger, int, int, Exception> _senderRetrying;
         private static readonly Action<ILogger, string, Exception> _exceptionOccuredWhileExecuting;
         private static readonly Action<ILogger, double, Exception> _messageHasBeenSent;
-        private static readonly Action<ILogger, int, Exception> _messagePublishException;
+        private static readonly Action<ILogger, int, string, Exception> _messagePublishException;
 
         static LoggerExtensions()
         {
@@ -60,10 +60,10 @@ namespace DotNetCore.CAP
                 "When call subscribe method, a parameter format conversion exception occurs. MethodName:'{MethodName}' ParameterName:'{ParameterName}' Content:'{Content}'."
             );
 
-            _senderRetrying = LoggerMessage.Define<int>(
+            _senderRetrying = LoggerMessage.Define<int, int>(
                 LogLevel.Debug,
                 3,
-                "Retrying send a message: {Retries}...");
+                "The {Retries}th retrying send a message failed. message id: {MessageId} ");
 
             _consumerExecuted = LoggerMessage.Define<double>(
                 LogLevel.Debug,
@@ -78,17 +78,17 @@ namespace DotNetCore.CAP
             _exceptionOccuredWhileExecuting = LoggerMessage.Define<string>(
                 LogLevel.Error,
                 6,
-                "An exception occured while trying to store a message: '{MessageId}'. ");
+                "An exception occured while trying to store a message. message id: {MessageId}");
 
             _messageHasBeenSent = LoggerMessage.Define<double>(
                 LogLevel.Debug,
                 4,
                 "Message published. Took: {Seconds} secs.");
 
-            _messagePublishException = LoggerMessage.Define<int>(
+            _messagePublishException = LoggerMessage.Define<int, string>(
                 LogLevel.Error,
                 6,
-                "An exception occured while publishing a message: '{MessageId}'. ");
+                "An exception occured while publishing a message, reason:{Reason}. message id:{MessageId}");
         }
 
         public static void ConsumerExecutionFailedWillRetry(this ILogger logger, Exception ex)
@@ -96,9 +96,9 @@ namespace DotNetCore.CAP
             _consumerFailedWillRetry(logger, ex);
         }
 
-        public static void SenderRetrying(this ILogger logger, int retries)
+        public static void SenderRetrying(this ILogger logger, int messageId, int retries)
         {
-            _senderRetrying(logger, retries, null);
+            _senderRetrying(logger, messageId, retries, null);
         }
 
         public static void MessageHasBeenSent(this ILogger logger, double seconds)
@@ -106,9 +106,9 @@ namespace DotNetCore.CAP
             _messageHasBeenSent(logger, seconds, null);
         }
 
-        public static void MessagePublishException(this ILogger logger, int messageId, Exception ex)
+        public static void MessagePublishException(this ILogger logger, int messageId, string reason, Exception ex)
         {
-            _messagePublishException(logger, messageId, ex);
+            _messagePublishException(logger, messageId, reason, ex);
         }
 
         public static void ConsumerExecuted(this ILogger logger, double seconds)

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

@@ -65,33 +65,46 @@ namespace DotNetCore.CAP.Processor
                     continue;
                 }
 
-                if (!hasException)
-                {
-                    try
-                    {
-                        _options.FailedCallback?.Invoke(MessageType.Publish, message.Name, message.Content);
-                    }
-                    catch (Exception ex)
-                    {
-                        hasException = true;
-                        _logger.LogWarning("Failed call-back method raised an exception:" + ex.Message);
-                    }
-                }
-
                 using (var transaction = connection.CreateTransaction())
                 {
-                    try
+                    var result = await _publishExecutor.PublishAsync(message.Name, message.Content);
+                    if (result.Succeeded)
                     {
-                        await _publishExecutor.PublishAsync(message.Name, message.Content);
-
                         _stateChanger.ChangeState(message, new SucceededState(), transaction);
+                        _logger.LogInformation("The message was sent successfully during the retry. MessageId:" + message.Id);
                     }
-                    catch (Exception e)
+                    else
                     {
-                        message.Content = Helper.AddExceptionProperty(message.Content, e);
+                        message.Content = Helper.AddExceptionProperty(message.Content, result.Exception);
+                        message.Retries++;
+                        if (message.StatusName == StatusName.Scheduled)
+                        {
+                            message.ExpiresAt = GetDueTime(message.Added, message.Retries);
+                            message.StatusName = StatusName.Failed;
+                        }
                         transaction.UpdateMessage(message);
-                    }
 
+                        if (message.Retries >= _options.FailedRetryCount)
+                        {
+                            _logger.LogError($"The message still sent failed after {_options.FailedRetryCount} retries. We will stop retrying the message. " +
+                                             "MessageId:" + message.Id);
+                            if (message.Retries == _options.FailedRetryCount)
+                            {
+                                if (!hasException)
+                                {
+                                    try
+                                    {
+                                        _options.FailedThresholdCallback?.Invoke(MessageType.Publish, message.Name, message.Content);
+                                    }
+                                    catch (Exception ex)
+                                    {
+                                        hasException = true;
+                                        _logger.LogWarning("Failed call-back method raised an exception:" + ex.Message);
+                                    }
+                                }
+                            }
+                        }
+                    }
                     await transaction.CommitAsync();
                 }
 
@@ -113,25 +126,60 @@ namespace DotNetCore.CAP.Processor
                     continue;
                 }
 
-                if (!hasException)
+                using (var transaction = connection.CreateTransaction())
                 {
-                    try
+                    var result = await _subscriberExecutor.ExecuteAsync(message);
+                    if (result.Succeeded)
                     {
-                        _options.FailedCallback?.Invoke(MessageType.Subscribe, message.Name, message.Content);
+                        _stateChanger.ChangeState(message, new SucceededState(), transaction);
+                        _logger.LogInformation("The message was execute successfully during the retry. MessageId:" + message.Id);
                     }
-                    catch (Exception ex)
+                    else
                     {
-                        hasException = true;
-                        _logger.LogWarning("Failed call-back method raised an exception:" + ex.Message);
+                        message.Content = Helper.AddExceptionProperty(message.Content, result.Exception);
+                        message.Retries++;
+                        if (message.StatusName == StatusName.Scheduled)
+                        {
+                            message.ExpiresAt = GetDueTime(message.Added, message.Retries);
+                            message.StatusName = StatusName.Failed;
+                        }
+                        transaction.UpdateMessage(message);
+
+                        if (message.Retries >= _options.FailedRetryCount)
+                        {
+                            _logger.LogError($"[Subscriber]The message still executed failed after {_options.FailedRetryCount} retries. " +
+                                             "We will stop retrying to execute the message. message id:" + message.Id);
+
+                            if (message.Retries == _options.FailedRetryCount)
+                            {
+                                if (!hasException)
+                                {
+                                    try
+                                    {
+                                        _options.FailedThresholdCallback?.Invoke(MessageType.Subscribe, message.Name, message.Content);
+                                    }
+                                    catch (Exception ex)
+                                    {
+                                        hasException = true;
+                                        _logger.LogWarning("Failed call-back method raised an exception:" + ex.Message);
+                                    }
+                                }
+                            }
+                        }
                     }
+                    await transaction.CommitAsync();
                 }
 
-                await _subscriberExecutor.ExecuteAsync(message);
-
                 context.ThrowIfStopping();
 
                 await context.WaitAsync(_delay);
             }
         }
+
+        public DateTime GetDueTime(DateTime addedTime, int retries)
+        {
+            var retryBehavior = RetryBehavior.DefaultRetry;
+            return addedTime.AddSeconds(retryBehavior.RetryIn(retries));
+        }
     }
 }
\ No newline at end of file