Commit 8485fb2f authored by Savorboard's avatar Savorboard

rename CapTransaction to Transaction

parent 2f6b3502
...@@ -2,6 +2,7 @@ ...@@ -2,6 +2,7 @@
using System.Threading.Tasks; using System.Threading.Tasks;
using DotNetCore.CAP; using DotNetCore.CAP;
using Microsoft.AspNetCore.Mvc; using Microsoft.AspNetCore.Mvc;
using MySql.Data.MySqlClient;
namespace Sample.RabbitMQ.MySql.Controllers namespace Sample.RabbitMQ.MySql.Controllers
{ {
...@@ -28,7 +29,17 @@ namespace Sample.RabbitMQ.MySql.Controllers ...@@ -28,7 +29,17 @@ namespace Sample.RabbitMQ.MySql.Controllers
[Route("~/publish2")] [Route("~/publish2")]
public IActionResult PublishMessage2() public IActionResult PublishMessage2()
{ {
_capBus.Publish("sample.kafka.sqlserver4", DateTime.Now); using (var connection = new MySqlConnection("Server=192.168.10.110;Database=testcap;UserId=root;Password=123123;"))
{
using (var transaction = connection.BeginAndJoinToTransaction(_capBus))
{
//your business code
_capBus.Publish("sample.rabbitmq.mysql", DateTime.Now);
transaction.Commit();
}
}
return Ok(); return Ok();
} }
...@@ -37,7 +48,7 @@ namespace Sample.RabbitMQ.MySql.Controllers ...@@ -37,7 +48,7 @@ namespace Sample.RabbitMQ.MySql.Controllers
public async Task<IActionResult> PublishMessageWithTransaction() public async Task<IActionResult> PublishMessageWithTransaction()
{ {
using (var trans = await _dbContext.Database.BeginTransactionAsync()) using (var trans = await _dbContext.Database.BeginTransactionAsync())
using (var capTrans = _capBus.CapTransaction.Begin(trans)) using (var capTrans = _capBus.Transaction.Begin(trans))
{ {
for (int i = 0; i < 10; i++) for (int i = 0; i < 10; i++)
{ {
...@@ -49,6 +60,8 @@ namespace Sample.RabbitMQ.MySql.Controllers ...@@ -49,6 +60,8 @@ namespace Sample.RabbitMQ.MySql.Controllers
return Ok(); return Ok();
} }
[NonAction] [NonAction]
[CapSubscribe("#.rabbitmq.mysql")] [CapSubscribe("#.rabbitmq.mysql")]
public void ReceiveMessage(DateTime time) public void ReceiveMessage(DateTime time)
......
...@@ -14,7 +14,7 @@ namespace DotNetCore.CAP.Abstractions ...@@ -14,7 +14,7 @@ namespace DotNetCore.CAP.Abstractions
{ {
public abstract class CapPublisherBase : ICapPublisher public abstract class CapPublisherBase : ICapPublisher
{ {
private readonly CapTransactionBase _capTransaction; private readonly CapTransactionBase _transaction;
private readonly IMessagePacker _msgPacker; private readonly IMessagePacker _msgPacker;
private readonly IContentSerializer _serializer; private readonly IContentSerializer _serializer;
...@@ -27,14 +27,14 @@ namespace DotNetCore.CAP.Abstractions ...@@ -27,14 +27,14 @@ namespace DotNetCore.CAP.Abstractions
protected CapPublisherBase(IServiceProvider service) protected CapPublisherBase(IServiceProvider service)
{ {
ServiceProvider = service; ServiceProvider = service;
_capTransaction = service.GetRequiredService<CapTransactionBase>(); _transaction = service.GetRequiredService<CapTransactionBase>();
_msgPacker = service.GetRequiredService<IMessagePacker>(); _msgPacker = service.GetRequiredService<IMessagePacker>();
_serializer = service.GetRequiredService<IContentSerializer>(); _serializer = service.GetRequiredService<IContentSerializer>();
} }
protected IServiceProvider ServiceProvider { get; } protected IServiceProvider ServiceProvider { get; }
public ICapTransaction CapTransaction => _capTransaction; public ICapTransaction Transaction => _transaction;
public void Publish<T>(string name, T contentObj, string callbackName = null) public void Publish<T>(string name, T contentObj, string callbackName = null)
{ {
...@@ -65,10 +65,10 @@ namespace DotNetCore.CAP.Abstractions ...@@ -65,10 +65,10 @@ namespace DotNetCore.CAP.Abstractions
protected async Task PublishAsyncInternal(CapPublishedMessage message) protected async Task PublishAsyncInternal(CapPublishedMessage message)
{ {
if (CapTransaction.DbTransaction == null) if (Transaction.DbTransaction == null)
{ {
NotUseTransaction = true; NotUseTransaction = true;
CapTransaction.DbTransaction = GetDbTransaction(); Transaction.DbTransaction = GetDbTransaction();
} }
Guid operationId = default(Guid); Guid operationId = default(Guid);
...@@ -77,15 +77,15 @@ namespace DotNetCore.CAP.Abstractions ...@@ -77,15 +77,15 @@ namespace DotNetCore.CAP.Abstractions
{ {
operationId = s_diagnosticListener.WritePublishMessageStoreBefore(message); operationId = s_diagnosticListener.WritePublishMessageStoreBefore(message);
await ExecuteAsync(message, CapTransaction); await ExecuteAsync(message, Transaction);
_capTransaction.AddToSent(message); _transaction.AddToSent(message);
s_diagnosticListener.WritePublishMessageStoreAfter(operationId, message); s_diagnosticListener.WritePublishMessageStoreAfter(operationId, message);
if (NotUseTransaction || CapTransaction.AutoCommit) if (NotUseTransaction || Transaction.AutoCommit)
{ {
_capTransaction.Commit(); _transaction.Commit();
} }
} }
catch (Exception e) catch (Exception e)
...@@ -96,9 +96,9 @@ namespace DotNetCore.CAP.Abstractions ...@@ -96,9 +96,9 @@ namespace DotNetCore.CAP.Abstractions
} }
finally finally
{ {
if (NotUseTransaction || CapTransaction.AutoCommit) if (NotUseTransaction || Transaction.AutoCommit)
{ {
_capTransaction.Dispose(); _transaction.Dispose();
} }
} }
} }
......
...@@ -11,7 +11,7 @@ namespace DotNetCore.CAP ...@@ -11,7 +11,7 @@ namespace DotNetCore.CAP
/// </summary> /// </summary>
public interface ICapPublisher public interface ICapPublisher
{ {
ICapTransaction CapTransaction { get; } ICapTransaction Transaction { get; }
/// <summary> /// <summary>
/// Asynchronous publish an object message. /// Asynchronous publish an object message.
......
...@@ -119,7 +119,7 @@ namespace DotNetCore.CAP.Test ...@@ -119,7 +119,7 @@ namespace DotNetCore.CAP.Test
private class MyProducerService : ICapPublisher private class MyProducerService : ICapPublisher
{ {
public ICapTransaction CapTransaction { get; } public ICapTransaction Transaction { get; }
public Task PublishAsync<T>(string name, T contentObj, string callbackName = null, public Task PublishAsync<T>(string name, T contentObj, string callbackName = null,
CancellationToken cancellationToken = default(CancellationToken)) CancellationToken cancellationToken = default(CancellationToken))
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment