add async support for CAPTransaction

src/DotNetCore.CAP.MySql/ICapTransaction.MySql.cs

@@ -3,6 +3,8 @@
 
 using System.Data;
 using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
 using Microsoft.EntityFrameworkCore.Infrastructure;
 using Microsoft.EntityFrameworkCore.Storage;
 using Microsoft.Extensions.DependencyInjection;
@@ -33,6 +35,22 @@ namespace DotNetCore.CAP
             Flush();
         }
 
+        public override async Task CommitAsync(CancellationToken cancellationToken = default)
+        {
+            Debug.Assert(DbTransaction != null);
+
+            switch (DbTransaction)
+            {
+                case IDbTransaction dbTransaction:
+                    dbTransaction.Commit();
+                    break;
+                case IDbContextTransaction dbContextTransaction:
+                    await dbContextTransaction.CommitAsync(cancellationToken);
+                    break;
+            }
+            Flush();
+        }
+
         public override void Rollback()
         {
             Debug.Assert(DbTransaction != null);
@@ -48,6 +66,21 @@ namespace DotNetCore.CAP
             }
         }
 
+        public override async Task RollbackAsync(CancellationToken cancellationToken = default)
+        {
+            Debug.Assert(DbTransaction != null);
+
+            switch (DbTransaction)
+            {
+                case IDbTransaction dbTransaction:
+                    dbTransaction.Rollback();
+                    break;
+                case IDbContextTransaction dbContextTransaction:
+                    await dbContextTransaction.RollbackAsync(cancellationToken);
+                    break;
+            }
+        }
+
         public override void Dispose()
         {
             (DbTransaction as IDbTransaction)?.Dispose();

src/DotNetCore.CAP.MySql/IDbContextTransaction.CAP.cs

@@ -2,6 +2,8 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
+using System.Threading;
+using System.Threading.Tasks;
 using DotNetCore.CAP;
 
 // ReSharper disable once CheckNamespace
@@ -15,10 +17,12 @@ namespace Microsoft.EntityFrameworkCore.Storage
         public CapEFDbTransaction(ICapTransaction transaction)
         {
             _transaction = transaction;
-            var dbContextTransaction = (IDbContextTransaction) _transaction.DbTransaction;
+            var dbContextTransaction = (IDbContextTransaction)_transaction.DbTransaction;
             TransactionId = dbContextTransaction.TransactionId;
         }
 
+        public Guid TransactionId { get; }
+
         public void Dispose()
         {
             _transaction.Dispose();
@@ -29,11 +33,25 @@ namespace Microsoft.EntityFrameworkCore.Storage
             _transaction.Commit();
         }
 
+        public Task CommitAsync(CancellationToken cancellationToken = default)
+        {
+            return _transaction.CommitAsync(cancellationToken);
+        }
+
         public void Rollback()
         {
             _transaction.Rollback();
         }
 
-        public Guid TransactionId { get; }
+        public Task RollbackAsync(CancellationToken cancellationToken = default)
+        {
+            return _transaction.CommitAsync(cancellationToken);
+        }
+
+        public ValueTask DisposeAsync()
+        {
+            Dispose();
+            return new ValueTask();
+        }
     }
 }

src/DotNetCore.CAP/ICapTransaction.Base.cs

@@ -1,4 +1,6 @@
 using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
 using DotNetCore.CAP.Persistence;
 
 namespace DotNetCore.CAP
@@ -36,8 +38,12 @@ namespace DotNetCore.CAP
 
         public abstract void Commit();
 
+        public abstract Task CommitAsync(CancellationToken cancellationToken = default);
+
         public abstract void Rollback();
 
+        public abstract Task RollbackAsync(CancellationToken cancellationToken = default);
+
         public abstract void Dispose();
     }
 }

src/DotNetCore.CAP/ICapTransaction.cs

@@ -2,6 +2,8 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace DotNetCore.CAP
 {
@@ -25,9 +27,13 @@ namespace DotNetCore.CAP
         /// </summary>
         void Commit();
 
+        Task CommitAsync(CancellationToken cancellationToken = default);
+
         /// <summary>
         /// We will delete the message data that has not been sstore in the buffer data of current transaction context.
         /// </summary>
         void Rollback();
+
+        Task RollbackAsync(CancellationToken cancellationToken = default);
     }
 }