refactor sql storage.

src/DotNetCore.CAP.MySql/MySqlStorage.cs

@@ -14,16 +14,20 @@ namespace DotNetCore.CAP.MySql
         private readonly IDbConnection _existingConnection = null;
         private readonly ILogger _logger;
         private readonly MySqlOptions _options;
+        private readonly CapOptions _capOptions;
 
-        public MySqlStorage(ILogger<MySqlStorage> logger, MySqlOptions options)
+        public MySqlStorage(ILogger<MySqlStorage> logger,
+            MySqlOptions options,
+            CapOptions capOptions)
         {
             _options = options;
+            _capOptions = capOptions;
             _logger = logger;
         }
 
         public IStorageConnection GetConnection()
         {
-            return new MySqlStorageConnection(_options);
+            return new MySqlStorageConnection(_options, _capOptions);
         }
 
         public IMonitoringApi GetMonitoringApi()

src/DotNetCore.CAP.MySql/MySqlStorageConnection.cs

@@ -11,12 +11,14 @@ namespace DotNetCore.CAP.MySql
 {
     public class MySqlStorageConnection : IStorageConnection
     {
+        private readonly CapOptions _capOptions;
         private readonly string _prefix;
 
         private const string DateTimeMaxValue = "9999-12-31 23:59:59";
 
-        public MySqlStorageConnection(MySqlOptions options)
+        public MySqlStorageConnection(MySqlOptions options, CapOptions capOptions)
         {
+            _capOptions = capOptions;
             Options = options;
             _prefix = Options.TableNamePrefix;
         }
@@ -43,9 +45,9 @@ namespace DotNetCore.CAP.MySql
             var sql = $@"
 SELECT `MessageId`,`MessageType` FROM `{_prefix}.queue` LIMIT 1 FOR UPDATE;
 DELETE FROM `{_prefix}.queue` LIMIT 1;";
-//            var sql = $@"
-//SELECT @MId:=`MessageId` as MessageId, @MType:=`MessageType` as MessageType FROM `{_prefix}.queue` LIMIT 1;
-//DELETE FROM `{_prefix}.queue` where `MessageId` = @MId AND `MessageType`=@MType;";
+            //            var sql = $@"
+            //SELECT @MId:=`MessageId` as MessageId, @MType:=`MessageType` as MessageType FROM `{_prefix}.queue` LIMIT 1;
+            //DELETE FROM `{_prefix}.queue` where `MessageId` = @MId AND `MessageType`=@MType;";
 
             return FetchNextMessageCoreAsync(sql);
         }
@@ -64,7 +66,7 @@ SELECT * FROM `{_prefix}.published` WHERE Id=LAST_INSERT_ID();";
 
         public async Task<IEnumerable<CapPublishedMessage>> GetFailedPublishedMessages()
         {
-            var sql = $"SELECT * FROM `{_prefix}.published` WHERE `StatusName` = '{StatusName.Failed}';";
+            var sql = $"SELECT * FROM `{_prefix}.published` WHERE `Retries`<{_capOptions.FailedRetryCount} AND `StatusName` = '{StatusName.Failed}' LIMIT 200;";
 
             using (var connection = new MySqlConnection(Options.ConnectionString))
             {
@@ -109,7 +111,7 @@ SELECT * FROM `{_prefix}.received` WHERE Id=LAST_INSERT_ID();";
 
         public async Task<IEnumerable<CapReceivedMessage>> GetFailedReceivedMessages()
         {
-            var sql = $"SELECT * FROM `{_prefix}.received` WHERE `StatusName` = '{StatusName.Failed}';";
+            var sql = $"SELECT * FROM `{_prefix}.received` WHERE `Retries`<{_capOptions.FailedRetryCount} AND `StatusName` = '{StatusName.Failed}' LIMIT 200;";
             using (var connection = new MySqlConnection(Options.ConnectionString))
             {
                 return await connection.QueryAsync<CapReceivedMessage>(sql);

src/DotNetCore.CAP.MySql/MySqlStorageTransaction.cs

@@ -29,7 +29,7 @@ namespace DotNetCore.CAP.MySql
             if (message == null) throw new ArgumentNullException(nameof(message));
 
             var sql =
-                $"UPDATE `{_prefix}.published` SET `Retries` = @Retries,`ExpiresAt` = @ExpiresAt,`StatusName`=@StatusName WHERE `Id`=@Id;";
+                $"UPDATE `{_prefix}.published` SET `Retries` = @Retries,`Content`= @Content,`ExpiresAt` = @ExpiresAt,`StatusName`=@StatusName WHERE `Id`=@Id;";
             _dbConnection.Execute(sql, message, _dbTransaction);
         }
 

src/DotNetCore.CAP.PostgreSql/PostgreSqlStorage.cs

@@ -13,17 +13,21 @@ namespace DotNetCore.CAP.PostgreSql
     {
         private readonly IDbConnection _existingConnection = null;
         private readonly ILogger _logger;
+        private readonly CapOptions _capOptions;
         private readonly PostgreSqlOptions _options;
 
-        public PostgreSqlStorage(ILogger<PostgreSqlStorage> logger, PostgreSqlOptions options)
+        public PostgreSqlStorage(ILogger<PostgreSqlStorage> logger,
+            CapOptions capOptions,
+            PostgreSqlOptions options)
         {
             _options = options;
             _logger = logger;
+            _capOptions = capOptions;
         }
 
         public IStorageConnection GetConnection()
         {
-            return new PostgreSqlStorageConnection(_options);
+            return new PostgreSqlStorageConnection(_options, _capOptions);
         }
 
         public IMonitoringApi GetMonitoringApi()

src/DotNetCore.CAP.PostgreSql/PostgreSqlStorageConnection.cs

@@ -11,8 +11,11 @@ namespace DotNetCore.CAP.PostgreSql
 {
     public class PostgreSqlStorageConnection : IStorageConnection
     {
-        public PostgreSqlStorageConnection(PostgreSqlOptions options)
+        private readonly CapOptions _capOptions;
+
+        public PostgreSqlStorageConnection(PostgreSqlOptions options,CapOptions capOptions)
         {
+            _capOptions = capOptions;
             Options = options;
         }
 
@@ -35,9 +38,7 @@ namespace DotNetCore.CAP.PostgreSql
 
         public Task<IFetchedMessage> FetchNextMessageAsync()
         {
-            var sql = $@"DELETE FROM ""{Options.Schema}"".""queue"" WHERE ""MessageId"" = (SELECT ""MessageId"" FROM ""{
-                    Options.Schema
-                }"".""queue"" FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING *;";
+            var sql = $@"DELETE FROM ""{Options.Schema}"".""queue"" WHERE ""MessageId"" = (SELECT ""MessageId"" FROM ""{Options.Schema}"".""queue"" FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING *;";
             return FetchNextMessageCoreAsync(sql);
         }
 
@@ -55,7 +56,7 @@ namespace DotNetCore.CAP.PostgreSql
         public async Task<IEnumerable<CapPublishedMessage>> GetFailedPublishedMessages()
         {
             var sql =
-                $"SELECT * FROM \"{Options.Schema}\".\"published\" WHERE \"StatusName\"='{StatusName.Failed}' LIMIT 1000;";
+                $"SELECT * FROM \"{Options.Schema}\".\"published\" WHERE \"Retries\"<{_capOptions.FailedRetryCount} AND \"StatusName\"='{StatusName.Failed}' LIMIT 200;";
 
             using (var connection = new NpgsqlConnection(Options.ConnectionString))
             {
@@ -98,7 +99,7 @@ namespace DotNetCore.CAP.PostgreSql
         public async Task<IEnumerable<CapReceivedMessage>> GetFailedReceivedMessages()
         {
             var sql =
-                $"SELECT * FROM \"{Options.Schema}\".\"received\" WHERE \"StatusName\"='{StatusName.Failed}' LIMIT 1000;";
+                $"SELECT * FROM \"{Options.Schema}\".\"received\" WHERE \"Retries\"<{_capOptions.FailedRetryCount} AND \"StatusName\"='{StatusName.Failed}' LIMIT 200;";
             using (var connection = new NpgsqlConnection(Options.ConnectionString))
             {
                 return await connection.QueryAsync<CapReceivedMessage>(sql);

src/DotNetCore.CAP.PostgreSql/PostgreSqlStorageTransaction.cs

@@ -31,7 +31,7 @@ namespace DotNetCore.CAP.PostgreSql
             var sql =
                 $@"UPDATE ""{
                         _schema
-                    }"".""published"" SET ""Retries""=@Retries,""ExpiresAt""=@ExpiresAt,""StatusName""=@StatusName WHERE ""Id""=@Id;";
+                    }"".""published"" SET ""Retries""=@Retries,""Content""= @Content,""ExpiresAt""=@ExpiresAt,""StatusName""=@StatusName WHERE ""Id""=@Id;";
             _dbConnection.Execute(sql, message, _dbTransaction);
         }
 

src/DotNetCore.CAP.SqlServer/SqlServerStorage.cs

@@ -13,17 +13,21 @@ namespace DotNetCore.CAP.SqlServer
     {
         private readonly IDbConnection _existingConnection = null;
         private readonly ILogger _logger;
+        private readonly CapOptions _capOptions;
         private readonly SqlServerOptions _options;
 
-        public SqlServerStorage(ILogger<SqlServerStorage> logger, SqlServerOptions options)
+        public SqlServerStorage(ILogger<SqlServerStorage> logger,
+            CapOptions capOptions,
+            SqlServerOptions options)
         {
             _options = options;
             _logger = logger;
+            _capOptions = capOptions;
         }
 
         public IStorageConnection GetConnection()
         {
-            return new SqlServerStorageConnection(_options);
+            return new SqlServerStorageConnection(_options, _capOptions);
         }
 
         public IMonitoringApi GetMonitoringApi()

src/DotNetCore.CAP.SqlServer/SqlServerStorageConnection.cs

@@ -11,8 +11,11 @@ namespace DotNetCore.CAP.SqlServer
 {
     public class SqlServerStorageConnection : IStorageConnection
     {
-        public SqlServerStorageConnection(SqlServerOptions options)
+        private readonly CapOptions _capOptions;
+
+        public SqlServerStorageConnection(SqlServerOptions options, CapOptions capOptions)
         {
+            _capOptions = capOptions;
             Options = options;
         }
 
@@ -57,7 +60,7 @@ OUTPUT DELETED.MessageId,DELETED.[MessageType];";
         public async Task<IEnumerable<CapPublishedMessage>> GetFailedPublishedMessages()
         {
             var sql =
-                $"SELECT * FROM [{Options.Schema}].[Published] WITH (readpast) WHERE StatusName = '{StatusName.Failed}'";
+                $"SELECT TOP (200) * FROM [{Options.Schema}].[Published] WITH (readpast) WHERE Retries<{_capOptions.FailedRetryCount} AND StatusName = '{StatusName.Failed}'";
 
             using (var connection = new SqlConnection(Options.ConnectionString))
             {
@@ -114,7 +117,7 @@ VALUES(@Name,@Group,@Content,@Retries,@Added,@ExpiresAt,@StatusName);";
         public async Task<IEnumerable<CapReceivedMessage>> GetFailedReceivedMessages()
         {
             var sql =
-                $"SELECT * FROM [{Options.Schema}].[Received] WITH (readpast) WHERE StatusName = '{StatusName.Failed}'";
+                $"SELECT TOP (200) * FROM [{Options.Schema}].[Received] WITH (readpast) WHERE Retries<{_capOptions.FailedRetryCount} AND StatusName = '{StatusName.Failed}'";
             using (var connection = new SqlConnection(Options.ConnectionString))
             {
                 return await connection.QueryAsync<CapReceivedMessage>(sql);

src/DotNetCore.CAP.SqlServer/SqlServerStorageTransaction.cs

@@ -29,7 +29,7 @@ namespace DotNetCore.CAP.SqlServer
             if (message == null) throw new ArgumentNullException(nameof(message));
 
             var sql =
-                $"UPDATE [{_schema}].[Published] SET [Retries] = @Retries,[ExpiresAt] = @ExpiresAt,[StatusName]=@StatusName WHERE Id=@Id;";
+                $"UPDATE [{_schema}].[Published] SET [Retries] = @Retries,[Content] = @Content,[ExpiresAt] = @ExpiresAt,[StatusName]=@StatusName WHERE Id=@Id;";
             _dbConnection.Execute(sql, message, _dbTransaction);
         }