Commit 1e250edd authored by Savorboard's avatar Savorboard

Cleanup Code

parent b55693b6
......@@ -13,8 +13,8 @@ namespace DotNetCore.CAP.MongoDB
{
public class MongoDBPublisher : CapPublisherBase, ICallbackPublisher
{
private readonly MongoDBOptions _options;
private readonly IMongoClient _client;
private readonly MongoDBOptions _options;
public MongoDBPublisher(IServiceProvider provider, MongoDBOptions options)
: base(provider)
......@@ -31,7 +31,7 @@ namespace DotNetCore.CAP.MongoDB
protected override Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
CancellationToken cancel = default(CancellationToken))
{
var insertOptions = new InsertOneOptions { BypassDocumentValidation = false };
var insertOptions = new InsertOneOptions {BypassDocumentValidation = false};
var collection = _client
.GetDatabase(_options.DatabaseName)
......@@ -41,7 +41,8 @@ namespace DotNetCore.CAP.MongoDB
{
return collection.InsertOneAsync(message, insertOptions, cancel);
}
var dbTrans = (IClientSessionHandle)transaction.DbTransaction;
var dbTrans = (IClientSessionHandle) transaction.DbTransaction;
return collection.InsertOneAsync(dbTrans, message, insertOptions, cancel);
}
}
......
......@@ -61,10 +61,10 @@ namespace DotNetCore.CAP
/// <summary>
/// Start the CAP transaction
/// </summary>
/// <param name="client">The <see cref="IMongoClient"/>.</param>
/// <param name="publisher">The <see cref="ICapPublisher"/>.</param>
/// <param name="client">The <see cref="IMongoClient" />.</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="IClientSessionHandle"/> of MongoDB transaction session object.</returns>
/// <returns>The <see cref="IClientSessionHandle" /> of MongoDB transaction session object.</returns>
public static IClientSessionHandle StartTransaction(this IMongoClient client,
ICapPublisher publisher, bool autoCommit = false)
{
......
......@@ -55,7 +55,8 @@ namespace DotNetCore.CAP.MongoDB
if (names.All(n => n != _options.PublishedCollection))
{
await database.CreateCollectionAsync(_options.PublishedCollection, cancellationToken: cancellationToken);
await database.CreateCollectionAsync(_options.PublishedCollection,
cancellationToken: cancellationToken);
}
_logger.LogDebug("Ensuring all create database tables script are applied.");
......
......@@ -47,7 +47,7 @@ namespace DotNetCore.CAP
using (var scope = x.CreateScope())
{
var provider = scope.ServiceProvider;
var dbContext = (DbContext)provider.GetService(mysqlOptions.DbContextType);
var dbContext = (DbContext) provider.GetService(mysqlOptions.DbContextType);
mysqlOptions.ConnectionString = dbContext.Database.GetDbConnection().ConnectionString;
return mysqlOptions;
}
......
// Copyright (c) .NET Core Community. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.
// ReSharper disable once CheckNamespace
namespace DotNetCore.CAP
{
public class MySqlOptions : EFOptions
......
......@@ -45,6 +45,7 @@ namespace DotNetCore.CAP.MySql
{
dbTrans = dbContextTrans.GetDbTransaction();
}
var conn = dbTrans?.Connection;
await conn.ExecuteAsync(PrepareSql(), message, dbTrans);
}
......
......@@ -67,7 +67,6 @@ namespace DotNetCore.CAP
public static ICapTransaction Begin(this ICapTransaction transaction,
IDbTransaction dbTransaction, bool autoCommit = false)
{
transaction.DbTransaction = dbTransaction;
transaction.AutoCommit = autoCommit;
......@@ -77,10 +76,10 @@ namespace DotNetCore.CAP
/// <summary>
/// Start the CAP transaction
/// </summary>
/// <param name="database">The <see cref="DatabaseFacade"/>.</param>
/// <param name="publisher">The <see cref="ICapPublisher"/>.</param>
/// <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)
{
......@@ -92,10 +91,10 @@ namespace DotNetCore.CAP
/// <summary>
/// Start the CAP transaction
/// </summary>
/// <param name="dbConnection">The <see cref="IDbConnection"/>.</param>
/// <param name="publisher">The <see cref="ICapPublisher"/>.</param>
/// <param name="dbConnection">The <see cref="IDbConnection" />.</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="ICapTransaction"/> object.</returns>
/// <returns>The <see cref="ICapTransaction" /> object.</returns>
public static ICapTransaction BeginTransaction(this IDbConnection dbConnection,
ICapPublisher publisher, bool autoCommit = false)
{
......
......@@ -126,7 +126,7 @@ select count(Id) from `{0}.received` where StatusName = N'Failed';", _prefix);
{
var sqlQuery = $"select count(Id) from `{_prefix}.{tableName}` where StatusName = @state";
var count = connection.ExecuteScalar<int>(sqlQuery, new { state = statusName });
var count = connection.ExecuteScalar<int>(sqlQuery, new {state = statusName});
return count;
}
......@@ -169,7 +169,7 @@ select aggr.* from (
var valuesMap = connection.Query<TimelineCounter>(
sqlQuery,
new { keys = keyMaps.Keys, statusName })
new {keys = keyMaps.Keys, statusName})
.ToDictionary(x => x.Key, x => x.Count);
foreach (var key in keyMaps.Keys)
......
// Copyright (c) .NET Core Community. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.
// ReSharper disable once CheckNamespace
namespace DotNetCore.CAP
{
public class PostgreSqlOptions : EFOptions
......
......@@ -17,6 +17,7 @@ namespace DotNetCore.CAP.PostgreSql
public class PostgreSqlPublisher : CapPublisherBase, ICallbackPublisher
{
private readonly PostgreSqlOptions _options;
public PostgreSqlPublisher(IServiceProvider provider) : base(provider)
{
_options = provider.GetService<PostgreSqlOptions>();
......@@ -44,6 +45,7 @@ namespace DotNetCore.CAP.PostgreSql
{
dbTrans = dbContextTrans.GetDbTransaction();
}
var conn = dbTrans?.Connection;
await conn.ExecuteAsync(PrepareSql(), message, dbTrans);
}
......@@ -62,6 +64,7 @@ namespace DotNetCore.CAP.PostgreSql
conn.Open();
return conn;
}
#endregion private methods
}
}
\ No newline at end of file
......@@ -76,10 +76,10 @@ namespace DotNetCore.CAP
/// <summary>
/// Start the CAP transaction
/// </summary>
/// <param name="dbConnection">The <see cref="IDbConnection"/>.</param>
/// <param name="publisher">The <see cref="ICapPublisher"/>.</param>
/// <param name="dbConnection">The <see cref="IDbConnection" />.</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="ICapTransaction"/> object.</returns>
/// <returns>The <see cref="ICapTransaction" /> object.</returns>
public static ICapTransaction BeginTransaction(this IDbConnection dbConnection,
ICapPublisher publisher, bool autoCommit = false)
{
......@@ -95,10 +95,10 @@ namespace DotNetCore.CAP
/// <summary>
/// Start the CAP transaction
/// </summary>
/// <param name="database">The <see cref="DatabaseFacade"/>.</param>
/// <param name="publisher">The <see cref="ICapPublisher"/>.</param>
/// <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)
{
......
......@@ -128,7 +128,7 @@ select count(""Id"") from ""{0}"".""received"" where ""StatusName"" = N'Failed'
var sqlQuery =
$"select count(\"Id\") from \"{_options.Schema}\".\"{tableName}\" where Lower(\"StatusName\") = Lower(@state)";
var count = connection.ExecuteScalar<int>(sqlQuery, new { state = statusName });
var count = connection.ExecuteScalar<int>(sqlQuery, new {state = statusName});
return count;
}
......@@ -170,7 +170,7 @@ with aggr as (
)
select ""Key"",""Count"" from aggr where ""Key""= Any(@keys);";
var valuesMap = connection.Query<TimelineCounter>(sqlQuery, new { keys = keyMaps.Keys.ToList(), statusName })
var valuesMap = connection.Query<TimelineCounter>(sqlQuery, new {keys = keyMaps.Keys.ToList(), statusName})
.ToList()
.ToDictionary(x => x.Key, x => x.Count);
......
......@@ -86,10 +86,6 @@ namespace DotNetCore.CAP.PostgreSql
}
}
public void Dispose()
{
}
public bool ChangePublishedState(long messageId, string state)
{
var sql =
......@@ -111,5 +107,9 @@ namespace DotNetCore.CAP.PostgreSql
return connection.Execute(sql) > 0;
}
}
public void Dispose()
{
}
}
}
\ No newline at end of file
......@@ -49,7 +49,7 @@ namespace DotNetCore.CAP
using (var scope = x.CreateScope())
{
var provider = scope.ServiceProvider;
var dbContext = (DbContext)provider.GetService(sqlServerOptions.DbContextType);
var dbContext = (DbContext) provider.GetService(sqlServerOptions.DbContextType);
sqlServerOptions.ConnectionString = dbContext.Database.GetDbConnection().ConnectionString;
return sqlServerOptions;
}
......
......@@ -12,8 +12,12 @@ namespace DotNetCore.CAP.SqlServer.Diagnostics
{
internal class DiagnosticObserver : IObserver<KeyValuePair<string, object>>
{
private readonly IDispatcher _dispatcher;
private const string SqlClientPrefix = "System.Data.SqlClient.";
public const string SqlAfterCommitTransaction = SqlClientPrefix + "WriteTransactionCommitAfter";
public const string SqlErrorCommitTransaction = SqlClientPrefix + "WriteTransactionCommitError";
private readonly ConcurrentDictionary<Guid, List<CapPublishedMessage>> _bufferList;
private readonly IDispatcher _dispatcher;
public DiagnosticObserver(IDispatcher dispatcher,
ConcurrentDictionary<Guid, List<CapPublishedMessage>> bufferList)
......@@ -22,19 +26,12 @@ namespace DotNetCore.CAP.SqlServer.Diagnostics
_bufferList = bufferList;
}
private const string SqlClientPrefix = "System.Data.SqlClient.";
public const string SqlAfterCommitTransaction = SqlClientPrefix + "WriteTransactionCommitAfter";
public const string SqlErrorCommitTransaction = SqlClientPrefix + "WriteTransactionCommitError";
public void OnCompleted()
{
}
public void OnError(Exception error)
{
}
public void OnNext(KeyValuePair<string, object> evt)
......@@ -60,7 +57,7 @@ namespace DotNetCore.CAP.SqlServer.Diagnostics
}
}
static object GetProperty(object _this, string propertyName)
private static object GetProperty(object _this, string propertyName)
{
return _this.GetType().GetTypeInfo().GetDeclaredProperty(propertyName)?.GetValue(_this);
}
......
......@@ -11,10 +11,8 @@ namespace DotNetCore.CAP.SqlServer.Diagnostics
{
public class DiagnosticProcessorObserver : IObserver<DiagnosticListener>
{
private readonly IDispatcher _dispatcher;
public const string DiagnosticListenerName = "SqlClientDiagnosticListener";
public ConcurrentDictionary<Guid, List<CapPublishedMessage>> BufferList { get; }
private readonly IDispatcher _dispatcher;
public DiagnosticProcessorObserver(IDispatcher dispatcher)
{
......@@ -22,6 +20,8 @@ namespace DotNetCore.CAP.SqlServer.Diagnostics
BufferList = new ConcurrentDictionary<Guid, List<CapPublishedMessage>>();
}
public ConcurrentDictionary<Guid, List<CapPublishedMessage>> BufferList { get; }
public void OnCompleted()
{
}
......
......@@ -59,6 +59,5 @@ namespace DotNetCore.CAP.SqlServer
}
#endregion private methods
}
}
\ No newline at end of file
......@@ -30,6 +30,7 @@ namespace DotNetCore.CAP
{
_dbContext = serviceProvider.GetService(sqlServerOptions.DbContextType) as DbContext;
}
_diagnosticProcessor = serviceProvider.GetRequiredService<DiagnosticProcessorObserver>();
}
......@@ -55,14 +56,14 @@ namespace DotNetCore.CAP
}
}
var transactionKey = ((SqlConnection)dbTransaction.Connection).ClientConnectionId;
var transactionKey = ((SqlConnection) dbTransaction.Connection).ClientConnectionId;
if (_diagnosticProcessor.BufferList.TryGetValue(transactionKey, out var list))
{
list.Add(msg);
}
else
{
var msgList = new List<CapPublishedMessage>(1) { msg };
var msgList = new List<CapPublishedMessage>(1) {msg};
_diagnosticProcessor.BufferList.TryAdd(transactionKey, msgList);
}
}
......@@ -134,10 +135,10 @@ namespace DotNetCore.CAP
/// <summary>
/// Start the CAP transaction
/// </summary>
/// <param name="dbConnection">The <see cref="IDbConnection"/>.</param>
/// <param name="publisher">The <see cref="ICapPublisher"/>.</param>
/// <param name="dbConnection">The <see cref="IDbConnection" />.</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="ICapTransaction"/> object.</returns>
/// <returns>The <see cref="ICapTransaction" /> object.</returns>
public static IDbTransaction BeginTransaction(this IDbConnection dbConnection,
ICapPublisher publisher, bool autoCommit = false)
{
......@@ -148,16 +149,16 @@ namespace DotNetCore.CAP
var dbTransaction = dbConnection.BeginTransaction();
var capTransaction = publisher.Transaction.Begin(dbTransaction, autoCommit);
return (IDbTransaction)capTransaction.DbTransaction;
return (IDbTransaction) capTransaction.DbTransaction;
}
/// <summary>
/// Start the CAP transaction
/// </summary>
/// <param name="database">The <see cref="DatabaseFacade"/>.</param>
/// <param name="publisher">The <see cref="ICapPublisher"/>.</param>
/// <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)
{
......
......@@ -135,7 +135,7 @@ select count(Id) from [{0}].Received with (nolock) where StatusName = N'Failed';
var sqlQuery =
$"select count(Id) from [{_options.Schema}].{tableName} with (nolock) where StatusName = @state";
var count = connection.ExecuteScalar<int>(sqlQuery, new { state = statusName });
var count = connection.ExecuteScalar<int>(sqlQuery, new {state = statusName});
return count;
}
......@@ -189,7 +189,7 @@ select [Key], [Count] from aggr with (nolock) where [Key] in @keys;";
var valuesMap = connection.Query<TimelineCounter>(
_options.IsSqlServer2008 ? sqlQuery2008 : sqlQuery,
new { keys = keyMaps.Keys, statusName })
new {keys = keyMaps.Keys, statusName})
.ToDictionary(x => x.Key, x => x.Count);
foreach (var key in keyMaps.Keys)
......
......@@ -17,10 +17,10 @@ namespace DotNetCore.CAP.SqlServer
public class SqlServerStorage : IStorage
{
private readonly CapOptions _capOptions;
private readonly DiagnosticProcessorObserver _diagnosticProcessorObserver;
private readonly IDbConnection _existingConnection = null;
private readonly ILogger _logger;
private readonly SqlServerOptions _options;
private readonly DiagnosticProcessorObserver _diagnosticProcessorObserver;
public SqlServerStorage(ILogger<SqlServerStorage> logger,
CapOptions capOptions,
......
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