Fix thread safe issue of ICapPublisher bug. (#371)

src/DotNetCore.CAP.MongoDB/ICapPublisher.MongoDB.cs

@@ -28,8 +28,9 @@ namespace DotNetCore.CAP.MongoDB
             await PublishAsyncInternal(message);
         }
 
-        protected override Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
-            CancellationToken cancel = default(CancellationToken))
+        protected override Task ExecuteAsync(CapPublishedMessage message,
+            ICapTransaction transaction = null,
+            CancellationToken cancel = default)
         {
             var insertOptions = new InsertOneOptions { BypassDocumentValidation = false };
 
@@ -49,7 +50,7 @@ namespace DotNetCore.CAP.MongoDB
                 Version = _options.Version,
             };
 
-            if (NotUseTransaction)
+            if (transaction == null)
             {
                 return collection.InsertOneAsync(store, insertOptions, cancel);
             }

src/DotNetCore.CAP.MongoDB/ICapTransaction.MongoDB.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System.Diagnostics;
+using Microsoft.Extensions.DependencyInjection;
 using MongoDB.Driver;
 
 // ReSharper disable once CheckNamespace
@@ -70,7 +71,8 @@ namespace DotNetCore.CAP
             ICapPublisher publisher, bool autoCommit = false)
         {
             var clientSessionHandle = client.StartSession();
-            var capTrans = publisher.Transaction.Begin(clientSessionHandle, autoCommit);
+            publisher.Transaction.Value = publisher.ServiceProvider.GetService<CapTransactionBase>();
+            var capTrans = publisher.Transaction.Value.Begin(clientSessionHandle, autoCommit);
             return new CapMongoDbClientSessionHandle(capTrans);
         }
     }

src/DotNetCore.CAP.MySql/ICapPublisher.MySql.cs

@@ -29,10 +29,11 @@ namespace DotNetCore.CAP.MySql
             await PublishAsyncInternal(message);
         }
 
-        protected override async Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
+        protected override async Task ExecuteAsync(CapPublishedMessage message,
+            ICapTransaction transaction = null,
             CancellationToken cancel = default(CancellationToken))
         {
-            if (NotUseTransaction)
+            if (transaction == null)
             {
                 using (var connection = new MySqlConnection(_options.ConnectionString))
                 {

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

@@ -3,15 +3,18 @@
 
 using System.Data;
 using System.Diagnostics;
+using System.Threading;
 using Microsoft.EntityFrameworkCore.Infrastructure;
 using Microsoft.EntityFrameworkCore.Storage;
+using Microsoft.Extensions.DependencyInjection;
 
 // ReSharper disable once CheckNamespace
 namespace DotNetCore.CAP
 {
     public class MySqlCapTransaction : CapTransactionBase
     {
-        public MySqlCapTransaction(IDispatcher dispatcher) : base(dispatcher)
+        public MySqlCapTransaction(
+            IDispatcher dispatcher) : base(dispatcher)
         {
         }
 
@@ -28,7 +31,6 @@ namespace DotNetCore.CAP
                     dbContextTransaction.Commit();
                     break;
             }
-
             Flush();
         }
 
@@ -85,7 +87,8 @@ namespace DotNetCore.CAP
             ICapPublisher publisher, bool autoCommit = false)
         {
             var trans = database.BeginTransaction();
-            var capTrans = publisher.Transaction.Begin(trans, autoCommit);
+            publisher.Transaction.Value = publisher.ServiceProvider.GetService<CapTransactionBase>();
+            var capTrans = publisher.Transaction.Value.Begin(trans, autoCommit);
             return new CapEFDbTransaction(capTrans);
         }
 
@@ -105,7 +108,8 @@ namespace DotNetCore.CAP
             }
 
             var dbTransaction = dbConnection.BeginTransaction();
-            return publisher.Transaction.Begin(dbTransaction, autoCommit);
+            publisher.Transaction.Value = publisher.ServiceProvider.GetService<CapTransactionBase>();
+            return publisher.Transaction.Value.Begin(dbTransaction, autoCommit);
         }
     }
 }
\ No newline at end of file

src/DotNetCore.CAP.PostgreSql/ICapPublisher.PostgreSql.cs

@@ -29,10 +29,10 @@ namespace DotNetCore.CAP.PostgreSql
             await PublishAsyncInternal(message);
         }
 
-        protected override async Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
+        protected override async Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction = null,
             CancellationToken cancel = default(CancellationToken))
         {
-            if (NotUseTransaction)
+            if (transaction == null)
             {
                 using (var connection = InitDbConnection())
                 {

src/DotNetCore.CAP.PostgreSql/ICapTransaction.PostgreSql.cs

@@ -5,6 +5,7 @@ using System.Data;
 using System.Diagnostics;
 using Microsoft.EntityFrameworkCore.Infrastructure;
 using Microsoft.EntityFrameworkCore.Storage;
+using Microsoft.Extensions.DependencyInjection;
 
 // ReSharper disable once CheckNamespace
 namespace DotNetCore.CAP
@@ -90,7 +91,8 @@ namespace DotNetCore.CAP
             }
 
             var dbTransaction = dbConnection.BeginTransaction();
-            return publisher.Transaction.Begin(dbTransaction, autoCommit);
+            publisher.Transaction.Value = publisher.ServiceProvider.GetService<CapTransactionBase>();
+            return publisher.Transaction.Value.Begin(dbTransaction, autoCommit);
         }
 
         /// <summary>
@@ -99,12 +101,13 @@ namespace DotNetCore.CAP
         /// <param name="database">The <see cref="DatabaseFacade" />.</param>
         /// <param name="publisher">The <see cref="ICapPublisher" />.</param>
         /// <param name="autoCommit">Whether the transaction is automatically committed when the message is published</param>
-        /// <returns>The <see cref="IDbContextTransaction" /> of EF dbcontext transaction object.</returns>
+        /// <returns>The <see cref="IDbContextTransaction" /> of EF DbContext transaction object.</returns>
         public static IDbContextTransaction BeginTransaction(this DatabaseFacade database,
             ICapPublisher publisher, bool autoCommit = false)
         {
             var trans = database.BeginTransaction();
-            var capTrans = publisher.Transaction.Begin(trans, autoCommit);
+            publisher.Transaction.Value = publisher.ServiceProvider.GetService<CapTransactionBase>();
+            var capTrans = publisher.Transaction.Value.Begin(trans, autoCommit);
             return new CapEFDbTransaction(capTrans);
         }
     }

src/DotNetCore.CAP.SqlServer/ICapPublisher.SqlServer.cs

@@ -29,10 +29,10 @@ namespace DotNetCore.CAP.SqlServer
             await PublishAsyncInternal(message);
         }
 
-        protected override async Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
+        protected override async Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction = null,
             CancellationToken cancel = default(CancellationToken))
         {
-            if (NotUseTransaction)
+            if (transaction == null)
             {
                 using (var connection = new SqlConnection(_options.ConnectionString))
                 {

src/DotNetCore.CAP.SqlServer/ICapTransaction.SqlServer.cs

@@ -150,7 +150,8 @@ namespace DotNetCore.CAP
             }
 
             var dbTransaction = dbConnection.BeginTransaction();
-            var capTransaction = publisher.Transaction.Begin(dbTransaction, autoCommit);
+            publisher.Transaction.Value = publisher.ServiceProvider.GetService<CapTransactionBase>();
+            var capTransaction = publisher.Transaction.Value.Begin(dbTransaction, autoCommit);
             return (IDbTransaction)capTransaction.DbTransaction;
         }
 
@@ -165,7 +166,8 @@ namespace DotNetCore.CAP
             ICapPublisher publisher, bool autoCommit = false)
         {
             var trans = database.BeginTransaction();
-            var capTrans = publisher.Transaction.Begin(trans, autoCommit);
+            publisher.Transaction.Value = publisher.ServiceProvider.GetService<CapTransactionBase>();
+            var capTrans = publisher.Transaction.Value.Begin(trans, autoCommit);
             return new CapEFDbTransaction(capTrans);
         }
     }

src/DotNetCore.CAP/Abstractions/CapPublisherBase.cs

@@ -7,7 +7,6 @@ using System.Threading;
 using System.Threading.Tasks;
 using DotNetCore.CAP.Diagnostics;
 using DotNetCore.CAP.Infrastructure;
-using DotNetCore.CAP.Internal;
 using DotNetCore.CAP.Models;
 using Microsoft.Extensions.DependencyInjection;
 
@@ -17,9 +16,7 @@ namespace DotNetCore.CAP.Abstractions
     {
         private readonly IMessagePacker _msgPacker;
         private readonly IContentSerializer _serializer;
-        private CapTransactionBase _transaction;
-
-        protected bool NotUseTransaction;
+        private readonly IDispatcher _dispatcher;
 
         // ReSharper disable once InconsistentNaming
         protected static readonly DiagnosticListener s_diagnosticListener =
@@ -28,13 +25,15 @@ namespace DotNetCore.CAP.Abstractions
         protected CapPublisherBase(IServiceProvider service)
         {
             ServiceProvider = service;
+            _dispatcher = service.GetRequiredService<IDispatcher>();
             _msgPacker = service.GetRequiredService<IMessagePacker>();
             _serializer = service.GetRequiredService<IContentSerializer>();
+            Transaction = new AsyncLocal<ICapTransaction>();
         }
 
-        protected IServiceProvider ServiceProvider { get; }
+        public IServiceProvider ServiceProvider { get; }
 
-        public ICapTransaction Transaction => _transaction ?? (_transaction = ServiceProvider.GetRequiredService<CapTransactionBase>());
+        public AsyncLocal<ICapTransaction> Transaction { get; }
 
         public void Publish<T>(string name, T contentObj, string callbackName = null)
         {
@@ -65,27 +64,33 @@ namespace DotNetCore.CAP.Abstractions
 
         protected async Task PublishAsyncInternal(CapPublishedMessage message)
         {
-            if (Transaction.DbTransaction == null)
-            {
-                NotUseTransaction = true;
-                Transaction.DbTransaction = new NoopTransaction();
-            }
-
-            Guid operationId = default(Guid);
+            var operationId = default(Guid);
 
             try
             {
                 operationId = s_diagnosticListener.WritePublishMessageStoreBefore(message);
 
-                await ExecuteAsync(message, Transaction);
+                if (Transaction.Value?.DbTransaction == null)
+                {
+                    await ExecuteAsync(message);
+
+                    s_diagnosticListener.WritePublishMessageStoreAfter(operationId, message);
+
+                    _dispatcher.EnqueueToPublish(message);
+                }
+                else
+                {
+                    var transaction = (CapTransactionBase)Transaction.Value;
 
-                _transaction.AddToSent(message);
+                    await ExecuteAsync(message, transaction);
 
                     s_diagnosticListener.WritePublishMessageStoreAfter(operationId, message);
                    
-                if (NotUseTransaction || Transaction.AutoCommit)
+                    transaction.AddToSent(message);
+                    if (transaction.AutoCommit)
                     {
-                    _transaction.Commit();
+                        transaction.Commit();
+                    }
                 }
             }
             catch (Exception e)
@@ -94,17 +99,10 @@ namespace DotNetCore.CAP.Abstractions
 
                 throw;
             }
-            finally
-            {
-                if (NotUseTransaction || Transaction.AutoCommit)
-                {
-                    _transaction?.Dispose();
-                }
-            }
         }
 
         protected abstract Task ExecuteAsync(CapPublishedMessage message,
-            ICapTransaction transaction,
+            ICapTransaction transaction = null,
             CancellationToken cancel = default(CancellationToken));
 
         protected virtual string Serialize<T>(T obj, string callbackName = null)

src/DotNetCore.CAP/ICapPublisher.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 System.Threading;
 using System.Threading.Tasks;
 
@@ -11,10 +12,12 @@ namespace DotNetCore.CAP
     /// </summary>
     public interface ICapPublisher
     {
+        IServiceProvider ServiceProvider { get; }
+
         /// <summary>
         /// CAP transaction context object
         /// </summary>
-        ICapTransaction Transaction { get; }
+        AsyncLocal<ICapTransaction> Transaction { get; }
 
         /// <summary>
         /// Asynchronous publish an object message.

test/DotNetCore.CAP.Test/CAP.BuilderTest.cs

@@ -120,7 +120,9 @@ namespace DotNetCore.CAP.Test
 
         private class MyProducerService : ICapPublisher
         {
-            public ICapTransaction Transaction { get; }
+            public IServiceProvider ServiceProvider { get; }
+
+            public AsyncLocal<ICapTransaction> Transaction { get; }
 
             public Task PublishAsync<T>(string name, T contentObj, string callbackName = null,
                 CancellationToken cancellationToken = default(CancellationToken))