Skip to content

C
CAP
Commit cbc145b7 authored by Savorboard's avatar Savorboard
Browse files
Options

Adjustments do not use transactions internally when a transaction is not... 

Adjustments do not use transactions internally when a transaction is not available to publish a message
parent 7d1fe75b
Show whitespace changes
Inline Side-by-side
Showing with 84 additions and 139 deletions
src/DotNetCore.CAP.MySql/ICapPublisher.MySql.cs
View file @ cbc145b7
......@@ -8,31 +8,19 @@ using System.Threading.Tasks;
using Dapper;
using DotNetCore.CAP.Abstractions;
using DotNetCore.CAP.Models;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Storage;
using Microsoft.Extensions.DependencyInjection;
using MySql.Data.MySqlClient;
namespace DotNetCore.CAP.MySql
{
public class MySqlPublisher : CapPublisherBase, ICallbackPublisher, IDisposable
public class MySqlPublisher : CapPublisherBase, ICallbackPublisher
{
private readonly DbContext _dbContext;
private readonly MySqlOptions _options;
private readonly bool _isUsingEF;
private MySqlConnection _connection;
public MySqlPublisher(IServiceProvider provider, MySqlOptions options) : base(provider)
{
_options = options;
if (_options.DbContextType == null)
public MySqlPublisher(IServiceProvider provider) : base(provider)
{
return;
}
_isUsingEF = true;
_dbContext = (DbContext)ServiceProvider.GetService(_options.DbContextType);
_options = provider.GetService<MySqlOptions>();
}
public async Task PublishCallbackAsync(CapPublishedMessage message)
......@@ -40,32 +28,26 @@ namespace DotNetCore.CAP.MySql
await PublishAsyncInternal(message);
}
protected override Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
protected override async Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
CancellationToken cancel = default(CancellationToken))
{
var dbTrans = transaction.DbTransaction as IDbTransaction;
if (dbTrans == null && transaction.DbTransaction is IDbContextTransaction dbContextTrans)
if (NotUseTransaction)
{
dbTrans = dbContextTrans.GetDbTransaction();
using (var connection = new MySqlConnection(_options.ConnectionString))
{
await connection.OpenAsync(cancel);
await connection.ExecuteAsync(PrepareSql(), message);
return;
}
var conn = dbTrans?.Connection;
return conn.ExecuteAsync(PrepareSql(), message, dbTrans);
}
protected override object GetDbTransaction()
{
if (_isUsingEF)
{
var dbContextTransaction = _dbContext.Database.CurrentTransaction;
if (dbContextTransaction == null)
var dbTrans = transaction.DbTransaction as IDbTransaction;
if (dbTrans == null && transaction.DbTransaction is IDbContextTransaction dbContextTrans)
{
return InitDbConnection();
}
return dbContextTransaction;
dbTrans = dbContextTrans.GetDbTransaction();
}
return InitDbConnection();
var conn = dbTrans?.Connection;
await conn.ExecuteAsync(PrepareSql(), message, dbTrans);
}
#region private methods
......@@ -76,19 +58,6 @@ namespace DotNetCore.CAP.MySql
$"INSERT INTO `{_options.TableNamePrefix}.published` (`Id`,`Name`,`Content`,`Retries`,`Added`,`ExpiresAt`,`StatusName`)VALUES(@Id,@Name,@Content,@Retries,@Added,@ExpiresAt,@StatusName);";
}
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();
}
}
}
\ No newline at end of file
src/DotNetCore.CAP.PostgreSql/ICapPublisher.PostgreSql.cs
View file @ cbc145b7
......@@ -8,31 +8,18 @@ using System.Threading.Tasks;
using Dapper;
using DotNetCore.CAP.Abstractions;
using DotNetCore.CAP.Models;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Storage;
using Microsoft.Extensions.DependencyInjection;
using Npgsql;
namespace DotNetCore.CAP.PostgreSql
{
public class PostgreSqlPublisher : CapPublisherBase, ICallbackPublisher
{
private readonly DbContext _dbContext;
private readonly PostgreSqlOptions _options;
private readonly bool _isUsingEF;
private NpgsqlConnection _connection;
public PostgreSqlPublisher(IServiceProvider provider, PostgreSqlOptions options): base(provider)
{
_options = options;
if (_options.DbContextType == null)
public PostgreSqlPublisher(IServiceProvider provider) : base(provider)
{
return;
}
_isUsingEF = true;
_dbContext = (DbContext)ServiceProvider.GetService(_options.DbContextType);
_options = provider.GetService< PostgreSqlOptions>();
}
public async Task PublishCallbackAsync(CapPublishedMessage message)
......@@ -43,6 +30,14 @@ namespace DotNetCore.CAP.PostgreSql
protected override Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
CancellationToken cancel = default(CancellationToken))
{
if (NotUseTransaction)
{
using (var connection = InitDbConnection())
{
return connection.ExecuteAsync(PrepareSql(), message);
}
}
var dbTrans = transaction.DbTransaction as IDbTransaction;
if (dbTrans == null && transaction.DbTransaction is IDbContextTransaction dbContextTrans)
{
......@@ -52,22 +47,6 @@ namespace DotNetCore.CAP.PostgreSql
return conn.ExecuteAsync(PrepareSql(), message, dbTrans);
}
protected override object GetDbTransaction()
{
if (_isUsingEF)
{
var dbContextTransaction = _dbContext.Database.CurrentTransaction;
if (dbContextTransaction == null)
{
return InitDbConnection();
}
return dbContextTransaction;
}
return InitDbConnection();
}
#region private methods
private string PrepareSql()
......@@ -76,18 +55,12 @@ namespace DotNetCore.CAP.PostgreSql
$"INSERT INTO \"{_options.Schema}\".\"published\" (\"Id\",\"Name\",\"Content\",\"Retries\",\"Added\",\"ExpiresAt\",\"StatusName\")VALUES(@Id,@Name,@Content,@Retries,@Added,@ExpiresAt,@StatusName);";
}
private IDbTransaction InitDbConnection()
private IDbConnection InitDbConnection()
{
_connection = new NpgsqlConnection(_options.ConnectionString);
_connection.Open();
return _connection.BeginTransaction(IsolationLevel.ReadCommitted);
var conn = new NpgsqlConnection(_options.ConnectionString);
conn.Open();
return conn;
}
#endregion private methods
public void Dispose()
{
_dbContext?.Dispose();
_connection?.Dispose();
}
}
}
\ No newline at end of file
src/DotNetCore.CAP.SqlServer/ICapPublisher.SqlServer.cs
View file @ cbc145b7
......@@ -8,31 +8,20 @@ using System.Threading;
using System.Threading.Tasks;
using Dapper;
using DotNetCore.CAP.Abstractions;
using DotNetCore.CAP.Internal;
using DotNetCore.CAP.Models;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Storage;
using Microsoft.Extensions.DependencyInjection;
namespace DotNetCore.CAP.SqlServer
{
public class SqlServerPublisher : CapPublisherBase, ICallbackPublisher, IDisposable
public class SqlServerPublisher : CapPublisherBase, ICallbackPublisher
{
private readonly DbContext _dbContext;
private readonly SqlServerOptions _options;
private readonly bool _isUsingEF;
private SqlConnection _connection;
public SqlServerPublisher(IServiceProvider provider, SqlServerOptions options) : base(provider)
{
_options = options;
if (_options.DbContextType == null)
public SqlServerPublisher(IServiceProvider provider) : base(provider)
{
return;
}
_isUsingEF = true;
_dbContext = (DbContext)ServiceProvider.GetService(_options.DbContextType);
_options = ServiceProvider.GetService<SqlServerOptions>();
}
public async Task PublishCallbackAsync(CapPublishedMessage message)
......@@ -40,54 +29,45 @@ namespace DotNetCore.CAP.SqlServer
await PublishAsyncInternal(message);
}
protected override Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
protected override async Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
CancellationToken cancel = default(CancellationToken))
{
var dbTrans = transaction.DbTransaction as IDbTransaction;
if (dbTrans == null && transaction.DbTransaction is IDbContextTransaction dbContextTrans)
if (NotUseTransaction)
{
dbTrans = dbContextTrans.GetDbTransaction();
using (var connection = new SqlConnection(_options.ConnectionString))
{
await connection.OpenAsync(cancel);
await connection.ExecuteAsync(PrepareSql(), message);
return;
}
var conn = dbTrans?.Connection;
return conn.ExecuteAsync(PrepareSql(), message, dbTrans);
}
protected override object GetDbTransaction()
{
if (_isUsingEF)
{
var dbContextTransaction = _dbContext.Database.CurrentTransaction;
if (dbContextTransaction == null)
var dbTrans = transaction.DbTransaction as IDbTransaction;
if (dbTrans == null && transaction.DbTransaction is IDbContextTransaction dbContextTrans)
{
return InitDbConnection();
}
return dbContextTransaction;
dbTrans = dbContextTrans.GetDbTransaction();
}
return InitDbConnection();
var conn = dbTrans?.Connection;
await conn.ExecuteAsync(PrepareSql(), message, dbTrans);
}
#region private methods
private string PrepareSql()
{
return
$"INSERT INTO {_options.Schema}.[Published] ([Id],[Name],[Content],[Retries],[Added],[ExpiresAt],[StatusName])VALUES(@Id,@Name,@Content,@Retries,@Added,@ExpiresAt,@StatusName);";
}
private IDbTransaction InitDbConnection()
{
_connection = new SqlConnection(_options.ConnectionString);
_connection.Open();
return _connection.BeginTransaction(IsolationLevel.ReadCommitted);
}
//private IDbConnection InitDbConnection()
//{
// var conn = ;
// conn.OpenAsync();
// return conn;
//}
#endregion private methods
public void Dispose()
{
_dbContext?.Dispose();
_connection?.Dispose();
}
}
}
\ No newline at end of file
src/DotNetCore.CAP.SqlServer/ICapTransaction.SqlServer.cs
View file @ cbc145b7
......@@ -5,6 +5,7 @@ using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using DotNetCore.CAP.Internal;
using DotNetCore.CAP.Models;
using DotNetCore.CAP.SqlServer.Diagnostics;
using Microsoft.EntityFrameworkCore;
......@@ -34,6 +35,12 @@ namespace DotNetCore.CAP
protected override void AddToSent(CapPublishedMessage msg)
{
if (DbTransaction is NoopTransaction)
{
base.AddToSent(msg);
return;
}
var dbTransaction = DbTransaction as IDbTransaction;
if (dbTransaction == null)
{
......@@ -64,6 +71,9 @@ namespace DotNetCore.CAP
{
switch (DbTransaction)
{
case NoopTransaction _:
Flush();
break;
case IDbTransaction dbTransaction:
dbTransaction.Commit();
break;
......
src/DotNetCore.CAP/Abstractions/CapPublisherBase.cs
View file @ cbc145b7
......@@ -7,6 +7,7 @@ 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;
......@@ -68,7 +69,7 @@ namespace DotNetCore.CAP.Abstractions
if (Transaction.DbTransaction == null)
{
NotUseTransaction = true;
Transaction.DbTransaction = GetDbTransaction();
Transaction.DbTransaction = new NoopTransaction();
}
Guid operationId = default(Guid);
......@@ -103,8 +104,6 @@ namespace DotNetCore.CAP.Abstractions
}
}
protected abstract object GetDbTransaction();
protected abstract Task ExecuteAsync(CapPublishedMessage message,
ICapTransaction transaction,
CancellationToken cancel = default(CancellationToken));
......
src/DotNetCore.CAP/Infrastructure/SnowflakeId.cs
View file @ cbc145b7
......@@ -23,6 +23,7 @@ namespace DotNetCore.CAP.Infrastructure
private static SnowflakeId _snowflakeId;
private readonly object _lock = new object();
private static readonly object s_lock = new object();
private long _lastTimestamp = -1L;
private SnowflakeId(long workerId, long datacenterId, long sequence = 0L)
......@@ -45,9 +46,12 @@ namespace DotNetCore.CAP.Infrastructure
public long Sequence { get; internal set; }
public static SnowflakeId Default(long datacenterId = 0)
{
lock (s_lock)
{
return _snowflakeId ?? (_snowflakeId = new SnowflakeId(AppDomain.CurrentDomain.Id, datacenterId));
}
}
public virtual long NextId()
{
......
src/DotNetCore.CAP/Internal/NoopTransaction.cs 0 → 100644
View file @ cbc145b7
// Copyright (c) .NET Core Community. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.
namespace DotNetCore.CAP.Internal
{
public class NoopTransaction
{
}
}
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