[break changes] modify api for fix #141

src/DotNetCore.CAP.MySql/CAP.MySqlCapOptionsExtension.cs

@@ -27,6 +27,7 @@ namespace DotNetCore.CAP
             services.AddScoped<ICapPublisher, CapPublisher>();
             services.AddScoped<ICallbackPublisher, CapPublisher>();
             services.AddTransient<ICollectProcessor, MySqlCollectProcessor>();
+            services.AddTransient<CapTransactionBase, MySqlCapTransaction>();
 
             AddSingletionMySqlOptions(services);
         }

src/DotNetCore.CAP.MySql/CapPublisher.cs

@@ -3,27 +3,27 @@
 
 using System;
 using System.Data;
+using System.Threading;
 using System.Threading.Tasks;
 using Dapper;
 using DotNetCore.CAP.Abstractions;
 using DotNetCore.CAP.Models;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.EntityFrameworkCore.Storage;
-using Microsoft.Extensions.Logging;
 using MySql.Data.MySqlClient;
 
 namespace DotNetCore.CAP.MySql
 {
-    public class CapPublisher : CapPublisherBase, ICallbackPublisher
+    public class CapPublisher : CapPublisherBase, ICallbackPublisher, IDisposable
     {
         private readonly DbContext _dbContext;
         private readonly MySqlOptions _options;
+        private readonly bool _isUsingEF;
 
-        public CapPublisher(ILogger<CapPublisher> logger, IDispatcher dispatcher, IServiceProvider provider,
-            MySqlOptions options)
-            : base(logger, dispatcher)
+        private MySqlConnection _connection;
+
+        public CapPublisher(IServiceProvider provider, MySqlOptions options) : base(provider)
         {
-            ServiceProvider = provider;
             _options = options;
 
             if (_options.DbContextType == null)
@@ -31,47 +31,41 @@ namespace DotNetCore.CAP.MySql
                 return;
             }
 
-            IsUsingEF = true;
-            _dbContext = (DbContext) ServiceProvider.GetService(_options.DbContextType);
+            _isUsingEF = true;
+            _dbContext = (DbContext)ServiceProvider.GetService(_options.DbContextType);
         }
 
         public async Task PublishCallbackAsync(CapPublishedMessage message)
         {
-            using (var conn = new MySqlConnection(_options.ConnectionString))
-            {
-                var id = await conn.ExecuteScalarAsync<int>(PrepareSql(), message);
-                message.Id = id;
-                Enqueue(message);
-            }
+            await PublishAsyncInternal(message);
         }
 
-        protected override void PrepareConnectionForEF()
+        protected override Task<int> ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
+            CancellationToken cancel = default(CancellationToken))
         {
-            DbConnection = _dbContext.Database.GetDbConnection();
-            var dbContextTransaction = _dbContext.Database.CurrentTransaction;
-            var dbTrans = dbContextTransaction?.GetDbTransaction();
-            //DbTransaction is dispose in original
-            if (dbTrans?.Connection == null)
+            var dbTrans = transaction.DbTransaction as IDbTransaction;
+            if (dbTrans == null && transaction.DbTransaction is IDbContextTransaction dbContextTrans)
             {
-                IsCapOpenedTrans = true;
-                dbContextTransaction?.Dispose();
-                dbContextTransaction = _dbContext.Database.BeginTransaction(IsolationLevel.ReadCommitted);
-                dbTrans = dbContextTransaction.GetDbTransaction();
+                dbTrans = dbContextTrans.GetDbTransaction();
             }
-
-            DbTransaction = dbTrans;
+            var conn = dbTrans?.Connection;
+            return conn.ExecuteScalarAsync<int>(PrepareSql(), message, dbTrans);
         }
 
-        protected override int Execute(IDbConnection dbConnection, IDbTransaction dbTransaction,
-            CapPublishedMessage message)
+        protected override object GetDbTransaction()
         {
-            return dbConnection.ExecuteScalar<int>(PrepareSql(), message, dbTransaction);
-        }
+            if (_isUsingEF)
+            {
+                var dbContextTransaction = _dbContext.Database.CurrentTransaction;
+                if (dbContextTransaction == null)
+                {
+                    return InitDbConnection();
+                }
 
-        protected override async Task<int> ExecuteAsync(IDbConnection dbConnection, IDbTransaction dbTransaction,
-            CapPublishedMessage message)
-        {
-            return await dbConnection.ExecuteScalarAsync<int>(PrepareSql(), message, dbTransaction);
+                return dbContextTransaction;
+            }
+
+            return InitDbConnection();
         }
 
         #region private methods
@@ -82,6 +76,19 @@ namespace DotNetCore.CAP.MySql
                 $"INSERT INTO `{_options.TableNamePrefix}.published` (`Name`,`Content`,`Retries`,`Added`,`ExpiresAt`,`StatusName`)VALUES(@Name,@Content,@Retries,@Added,@ExpiresAt,@StatusName);SELECT LAST_INSERT_ID()";
         }
 
+        private IDbTransaction InitDbConnection()
+        {
+            _connection = new MySqlConnection(_options.ConnectionString);
+            _connection.Open();
+            return _connection.BeginTransaction(IsolationLevel.ReadCommitted);
+        }
+
         #endregion private methods
+
+        public void Dispose()
+        {
+            _dbContext?.Dispose();
+            _connection?.Dispose();
+        }
     }
 }

src/DotNetCore.CAP/DotNetCore.CAP.csproj

@@ -37,7 +37,6 @@
     <PackageReference Include="Microsoft.Extensions.Options" Version="2.1.0" />
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="2.1.0" />
     <PackageReference Include="Newtonsoft.Json" Version="11.0.2" />
-    <PackageReference Include="System.Data.Common" Version="4.3.0" />
     <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="4.5.0" />
     <PackageReference Include="System.Threading.ThreadPool" Version="4.3.0" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="2.1.0" />

src/DotNetCore.CAP/ICapPublisher.cs

@@ -1,9 +1,8 @@
 // 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.Data;
+using System.Threading;
 using System.Threading.Tasks;
-using DotNetCore.CAP.Abstractions;
 
 namespace DotNetCore.CAP
 {
@@ -12,66 +11,23 @@ namespace DotNetCore.CAP
     /// </summary>
     public interface ICapPublisher
     {
-        /// <summary>
-        /// (EntityFramework) Asynchronous publish an object message.
-        /// <para>
-        /// If you are using the EntityFramework, you need to configure the DbContextType first.
-        /// otherwise you need to use overloaded method with  IDbTransaction.
-        /// </para>
-        /// </summary>
-        /// <typeparam name="T">The type of content object.</typeparam>
-        /// <param name="name">the topic name or exchange router key.</param>
-        /// <param name="contentObj">message body content, that will be serialized of json.</param>
-        /// <param name="callbackName">callback subscriber name</param>
-        Task PublishAsync<T>(string name, T contentObj, string callbackName = null);
+        ICapTransaction CapTransaction { get; }
 
         /// <summary>
-        /// (EntityFramework) Publish an object message.
-        /// <para>
-        /// If you are using the EntityFramework, you need to configure the DbContextType first.
-        /// otherwise you need to use overloaded method with IDbTransaction.
-        /// </para>
+        /// Asynchronous publish an object message.
         /// </summary>
-        /// <typeparam name="T">The type of content object.</typeparam>
         /// <param name="name">the topic name or exchange router key.</param>
         /// <param name="contentObj">message body content, that will be serialized of json.</param>
         /// <param name="callbackName">callback subscriber name</param>
-        void Publish<T>(string name, T contentObj, string callbackName = null);
+        /// <param name="cancellationToken"></param>
+        Task PublishAsync<T>(string name, T contentObj, string callbackName = null, CancellationToken cancellationToken = default(CancellationToken));
 
         /// <summary>
-        /// (ado.net) Asynchronous publish an object message.
+        /// Publish an object message.
         /// </summary>
         /// <param name="name">the topic name or exchange router key.</param>
         /// <param name="contentObj">message body content, that will be serialized of json.</param>
-        /// <param name="dbTransaction">the transaction of <see cref="IDbTransaction" /></param>
         /// <param name="callbackName">callback subscriber name</param>
-        Task PublishAsync<T>(string name, T contentObj, IDbTransaction dbTransaction, string callbackName = null);
-
-        /// <summary>
-        /// (ado.net) Publish an object message.
-        /// </summary>
-        /// <param name="name">the topic name or exchange router key.</param>
-        /// <param name="contentObj">message body content, that will be serialized of json.</param>
-        /// <param name="dbTransaction">the transaction of <see cref="IDbTransaction" /></param>
-        /// <param name="callbackName">callback subscriber name</param>
-        void Publish<T>(string name, T contentObj, IDbTransaction dbTransaction, string callbackName = null);
-
-        /// <summary>
-        /// Publish an object message with mongo.
-        /// </summary>
-        /// <param name="name">the topic name or exchange router key.</param>
-        /// <param name="contentObj">message body content, that will be serialized of json.</param>
-        /// <param name="mongoTransaction">if transaction was set null, the message will be published directly.</param>
-        /// <param name="callbackName">callback subscriber name</param>
-        void PublishWithMongo<T>(string name, T contentObj, IMongoTransaction mongoTransaction = null, string callbackName = null);
-
-        /// <summary>
-        ///  Asynchronous publish an object message with mongo.
-        /// </summary>
-        /// <param name="name">the topic name or exchange router key.</param>
-        /// <param name="contentObj">message body content, that will be serialized of json.</param>
-        /// <param name="mongoTransaction">if transaction was set null, the message will be published directly.</param>
-        /// <param name="callbackName">callback subscriber name</param>
-        Task PublishWithMongoAsync<T>(string name, T contentObj, IMongoTransaction mongoTransaction = null, string callbackName = null);
+        void Publish<T>(string name, T contentObj, string callbackName = null);
     }
 }

src/DotNetCore.CAP/IPublishMessageSender.Base.cs

@@ -173,8 +173,6 @@ namespace DotNetCore.CAP
                 du);
 
             s_diagnosticListener.WritePublishAfter(eventData);
-
-            _logger.MessageHasBeenSent(du.TotalSeconds);
         }
 
         private void TracingError(Guid operationId, CapPublishedMessage message, OperateResult result, DateTimeOffset startTime, TimeSpan du)