modify fetched message method without delete queue message when get q…

…ueue message.
dotnetcore · Dec 17, 2017 · 3f4f349 · 3f4f349
1 parent 6d5c771
commit 3f4f349
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 10 deletions.
diff --git a/src/DotNetCore.CAP.MySql/MySqlFetchedMessage.cs b/src/DotNetCore.CAP.MySql/MySqlFetchedMessage.cs
@@ -7,12 +7,14 @@ namespace DotNetCore.CAP.MySql
     public class MySqlFetchedMessage : IFetchedMessage
     {
         private readonly MySqlOptions _options;
+        private readonly string _processId;
 
-        public MySqlFetchedMessage(int messageId, MessageType type, MySqlOptions options)
+        public MySqlFetchedMessage(int messageId, MessageType type, string processId, MySqlOptions options)
         {
             MessageId = messageId;
             MessageType = type;
 
+            _processId = processId;
             _options = options;
         }
 
@@ -22,15 +24,19 @@ public MySqlFetchedMessage(int messageId, MessageType type, MySqlOptions options
 
         public void RemoveFromQueue()
         {
-            // ignored
+            using (var connection = new MySqlConnection(_options.ConnectionString))
+            {
+                connection.Execute($"DELETE FROM `{_options.TableNamePrefix}.queue` WHERE `ProcessId`=@ProcessId"
+                    , new { ProcessId = _processId });
+            } 
         }
 
         public void Requeue()
         {
             using (var connection = new MySqlConnection(_options.ConnectionString))
             {
-                connection.Execute($"insert into `{_options.TableNamePrefix}.queue`(`MessageId`,`MessageType`) values(@MessageId,@MessageType);"
-                    , new {MessageId, MessageType });
+                connection.Execute($"UPDATE `{_options.TableNamePrefix}.queue` SET `ProcessId`=NULL WHERE `ProcessId`=@ProcessId"
+                    , new { ProcessId = _processId });
             }
         }
 

diff --git a/src/DotNetCore.CAP.MySql/MySqlStorageConnection.cs b/src/DotNetCore.CAP.MySql/MySqlStorageConnection.cs
@@ -41,12 +41,12 @@ public async Task<CapPublishedMessage> GetPublishedMessageAsync(int id)
 
         public Task<IFetchedMessage> FetchNextMessageAsync()
         {
+            var processId = ObjectId.GenerateNewStringId();
             var sql = $@"
 UPDATE `{_prefix}.queue` SET `ProcessId`=@ProcessId WHERE `ProcessId` IS NULL LIMIT 1;
-SELECT `MessageId`,`MessageType` FROM `{_prefix}.queue` WHERE `ProcessId`=@ProcessId;
-DELETE FROM `{_prefix}.queue` WHERE `ProcessId`=@ProcessId";
+SELECT `MessageId`,`MessageType` FROM `{_prefix}.queue` WHERE `ProcessId`=@ProcessId;";
 
-            return FetchNextMessageCoreAsync(sql, new { ProcessId = Guid.NewGuid().ToString() });
+            return FetchNextMessageCoreAsync(sql, processId);
         }
 
         public async Task<CapPublishedMessage> GetNextPublishedMessageToBeEnqueuedAsync()
@@ -139,18 +139,18 @@ public bool ChangeReceivedState(int messageId, string state)
             }
         }
 
-        private async Task<IFetchedMessage> FetchNextMessageCoreAsync(string sql, object args = null)
+        private async Task<IFetchedMessage> FetchNextMessageCoreAsync(string sql, string processId)
         {
             FetchedMessage fetchedMessage;
             using (var connection = new MySqlConnection(Options.ConnectionString))
             {
-                fetchedMessage = await connection.QuerySingleOrDefaultAsync<FetchedMessage>(sql, args);
+                fetchedMessage = await connection.QuerySingleOrDefaultAsync<FetchedMessage>(sql, new { ProcessId = processId });
             }
 
             if (fetchedMessage == null)
                 return null;
 
-            return new MySqlFetchedMessage(fetchedMessage.MessageId, fetchedMessage.MessageType, Options);
+            return new MySqlFetchedMessage(fetchedMessage.MessageId, fetchedMessage.MessageType, processId, Options);
         }
 
         public void Dispose()