Sent async message in the loop causes an exception

[yang-xiaodong]

Async publish message in controller or services with the loop :

        [Route("~/publish")]
        public async Task<IActionResult> PublishMessage()
        {
            for (int i = 0; i < 10; i++)
            {
                 await _capBus.PublishAsync("sample.rabbitmq.mysql", DateTime.Now);
            }
            return Ok();
        }

This type of sent can lead to an exception !

[ajdwfnhaps]

Thanks, Does it mean had fixed this bug at the latest nuget package version:2.1.1-preview-28414190 ?

[yang-xiaodong]

@ajdwfnhaps Yes, you can do a test !

[yang-xiaodong]

Version 2.1.1 has been released and fixed this bug.

[ajdwfnhaps]

Got it. Great

[ajdwfnhaps]

使用v2.1.1，现在没有在循环里，但异步发送MQ消息，偶尔还会报错

//遇到结束时，发送消息通知Quartz 动态启动用户汇总统计Job操作，并退出
            if (msg.eof == 1)
            {
                await _capBus.PublishAsync(MqOptions.AccountPeriodMacDetailEOFRouteKey, msg);
                return;
            }

[2017-11-30 10:50:00.7802] [] [WARN ] [DotNetCore.CAP.SubscribeQueueExecutor]:
Job failed to execute. Will retry.
System.InvalidOperationException: There is already an open DataReader associated with this Connection which must be closed first.
at MySqlConnector.Core.ServerSession.StartQuerying(MySqlCommand command) in C:\projects\mysqlconnector\src\MySqlConnector\Core\ServerSession.cs:line 115
at MySqlConnector.Core.TextCommandExecutor.d__3.MoveNext() in C:\projects\mysqlconnector\src\MySqlConnector\Core\TextCommandExecutor.cs:line 66
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at MySqlConnector.Core.TextCommandExecutor.d__1.MoveNext() in C:\projects\mysqlconnector\src\MySqlConnector\Core\TextCommandExecutor.cs:line 25
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at MySql.Data.MySqlClient.MySqlTransaction.d__2.MoveNext() in C:\projects\mysqlconnector\src\MySqlConnector\MySql.Data.MySqlClient\MySqlTransaction.cs:line 27
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at MySql.Data.MySqlClient.MySqlTransaction.Commit() in C:\projects\mysqlconnector\src\MySqlConnector\MySql.Data.MySqlClient\MySqlTransaction.cs:line 13
at DotNetCore.CAP.Abstractions.CapPublisherBase.ClosedCap() in D:\Work\aimei_finance\trunk\code\aimei_finance\CAP\DotNetCore.CAP\Abstractions\CapPublisherBase.cs:line 162
at DotNetCore.CAP.Abstractions.CapPublisherBase.d__35.MoveNext() in D:\Work\aimei_finance\trunk\code\aimei_finance\CAP\DotNetCore.CAP\Abstractions\CapPublisherBase.cs:line 137
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at AAM.Finance.Service.CAP.AccPeriodService.d__4.MoveNext() in D:\Work\aimei_finance\trunk\code\aimei_finance\CAP.Instance\AAM.Finance.Service\CAP\AccPeriodService.cs:line 48
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at lambda_method(Closure , Object )
at Microsoft.Extensions.Internal.ObjectMethodExecutorAwaitable.Awaiter.GetResult() in D:\Work\aimei_finance\trunk\code\aimei_finance\CAP\DotNetCore.CAP\Internal\ObjectMethodExecutor\ObjectMethodExecutorAwaitable.cs:line 90
at DotNetCore.CAP.Internal.DefaultConsumerInvoker.d__7.MoveNext() in D:\Work\aimei_finance\trunk\code\aimei_finance\CAP\DotNetCore.CAP\Internal\IConsumerInvoker.Default.cs:line 72
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Runtime.CompilerServices.TaskAwaiter1.GetResult() at DotNetCore.CAP.Internal.DefaultConsumerInvoker.<InvokeAsync>d__5.MoveNext() in D:\Work\aimei_finance\trunk\code\aimei_finance\CAP\DotNetCore.CAP\Internal\IConsumerInvoker.Default.cs:line 47 --- End of stack trace from previous location where exception was thrown --- at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw() at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) at System.Runtime.CompilerServices.TaskAwaiter1.GetResult()
at DotNetCore.CAP.Internal.DefaultSubscriberExecutor.d__7.MoveNext() in D:\Work\aimei_finance\trunk\code\aimei_finance\CAP\DotNetCore.CAP\Internal\ISubscriberExecutor.Default.cs:line 42

[yang-xiaodong]

@ajdwfnhaps 你这个可能是由于DI生命周期的问题，改成这样试一下

public class AccountPeriodJob : JobBase
{
    private readonly ILogger _logger;
    private readonly IServiceProvider _serviceProvider;

    public AccountPeriodJob(ILogger<AccountPeriodJob> logger, IServiceProvider serviceProvider)
    {
        _logger = logger;
        _serviceProvider = serviceProvider;
    }

    protected override async Task Invoke(IJobExecutionContext context)
    {
        using (var scope = _serviceProvider.CreateScope())
        {
            var provider = scope.ServiceProvider;
            var accountPeriodService = provider.GetRequiredService<IAccountPeriodService>();
            await accountPeriodService.SendAccountPeriodMQMessageAsync();
        }
        _logger.LogInformation("AccountPeriodJob Runing...");
    }
}

[ajdwfnhaps]

谢谢你的建议，我按照你提供的方法试了一下，问题仍然存在，但如果我换成同步发送（_capBus.Publish），就不会出现这个问题