Skip to content

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

ICapPublisher add synchronous publish methods.

parent 1c03de4f
Hide whitespace changes
Inline Side-by-side
Showing with 185 additions and 34 deletions
src/DotNetCore.CAP.SqlServer/CapPublisher.cs
View file @ 5842dfd5
...@@ -7,11 +7,13 @@ using DotNetCore.CAP.Models; ...@@ -7,11 +7,13 @@ using DotNetCore.CAP.Models;
using DotNetCore.CAP.Processor; using DotNetCore.CAP.Processor;
using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Storage; using Microsoft.EntityFrameworkCore.Storage;
using Microsoft.Extensions.Logging;
namespace DotNetCore.CAP.SqlServer namespace DotNetCore.CAP.SqlServer
{ {
public class CapPublisher : ICapPublisher public class CapPublisher : ICapPublisher
{ {
private readonly ILogger _logger;
private readonly SqlServerOptions _options; private readonly SqlServerOptions _options;
private readonly DbContext _dbContext; private readonly DbContext _dbContext;
...@@ -21,9 +23,12 @@ namespace DotNetCore.CAP.SqlServer ...@@ -21,9 +23,12 @@ namespace DotNetCore.CAP.SqlServer
protected IServiceProvider ServiceProvider { get; } protected IServiceProvider ServiceProvider { get; }
public CapPublisher(IServiceProvider provider, SqlServerOptions options) public CapPublisher(IServiceProvider provider,
ILogger<CapPublisher> logger,
SqlServerOptions options)
{ {
ServiceProvider = provider; ServiceProvider = provider;
_logger = logger;
_options = options; _options = options;
if (_options.DbContextType != null) if (_options.DbContextType != null)
...@@ -33,57 +38,152 @@ namespace DotNetCore.CAP.SqlServer ...@@ -33,57 +38,152 @@ namespace DotNetCore.CAP.SqlServer
} }
} }
public void Publish(string name, string content)
{
CheckIsUsingEF(name);
PublishCore(name, content);
}
public Task PublishAsync(string name, string content) public Task PublishAsync(string name, string content)
{ {
if (name == null) throw new ArgumentNullException(nameof(name)); CheckIsUsingEF(name);
if (!IsUsingEF) throw new InvalidOperationException("If you are using the EntityFramework, you need to configure the DbContextType first." +
" otherwise you need to use overloaded method with IDbConnection and IDbTransaction.");
return Publish(name, content); return PublishCoreAsync(name, content);
}
public void Publish<T>(string name, T contentObj)
{
CheckIsUsingEF(name);
var content = Helper.ToJson(contentObj);
PublishCore(name, content);
} }
public Task PublishAsync<T>(string name, T contentObj) public Task PublishAsync<T>(string name, T contentObj)
{ {
if (name == null) throw new ArgumentNullException(nameof(name)); CheckIsUsingEF(name);
if (!IsUsingEF) throw new InvalidOperationException("If you are using the EntityFramework, you need to configure the DbContextType first." +
" otherwise you need to use overloaded method with IDbConnection and IDbTransaction.");
var content = Helper.ToJson(contentObj); var content = Helper.ToJson(contentObj);
return Publish(name, content);
return PublishCoreAsync(name, content);
} }
public Task PublishAsync(string name, string content, IDbConnection dbConnection) public void Publish(string name, string content, IDbConnection dbConnection, IDbTransaction dbTransaction = null)
{ {
if (IsUsingEF) throw new InvalidOperationException("If you are using the EntityFramework, you do not need to use this overloaded."); CheckIsAdoNet(name);
if (name == null) throw new ArgumentNullException(nameof(name));
if (dbConnection == null) throw new ArgumentNullException(nameof(dbConnection)); if (dbConnection == null)
throw new ArgumentNullException(nameof(dbConnection));
var dbTransaction = dbConnection.BeginTransaction(IsolationLevel.ReadCommitted); dbTransaction = dbTransaction ?? dbConnection.BeginTransaction(IsolationLevel.ReadCommitted);
IsCapOpenedTrans = true; IsCapOpenedTrans = true;
return PublishWithTrans(name, content, dbConnection, dbTransaction);
PublishWithTrans(name, content, dbConnection, dbTransaction);
} }
public Task PublishAsync(string name, string content, IDbConnection dbConnection, IDbTransaction dbTransaction) public Task PublishAsync(string name, string content, IDbConnection dbConnection, IDbTransaction dbTransaction = null)
{
CheckIsAdoNet(name);
if (dbConnection == null)
throw new ArgumentNullException(nameof(dbConnection));
dbTransaction = dbTransaction ?? dbConnection.BeginTransaction(IsolationLevel.ReadCommitted);
IsCapOpenedTrans = true;
return PublishWithTransAsync(name, content, dbConnection, dbTransaction);
}
public void Publish<T>(string name, T contentObj, IDbConnection dbConnection, IDbTransaction dbTransaction = null)
{
CheckIsAdoNet(name);
if (dbConnection == null)
throw new ArgumentNullException(nameof(dbConnection));
var content = Helper.ToJson(contentObj);
dbTransaction = dbTransaction ?? dbConnection.BeginTransaction(IsolationLevel.ReadCommitted);
PublishWithTrans(name, content, dbConnection, dbTransaction);
}
public Task PublishAsync<T>(string name, T contentObj, IDbConnection dbConnection, IDbTransaction dbTransaction = null)
{
CheckIsAdoNet(name);
if (dbConnection == null)
throw new ArgumentNullException(nameof(dbConnection));
var content = Helper.ToJson(contentObj);
dbTransaction = dbTransaction ?? dbConnection.BeginTransaction(IsolationLevel.ReadCommitted);
return PublishWithTransAsync(name, content, dbConnection, dbTransaction);
}
#region private methods
private void CheckIsUsingEF(string name)
{ {
if (IsUsingEF) throw new InvalidOperationException("If you are using the EntityFramework, you do not need to use this overloaded.");
if (name == null) throw new ArgumentNullException(nameof(name)); if (name == null) throw new ArgumentNullException(nameof(name));
if (dbConnection == null) throw new ArgumentNullException(nameof(dbConnection)); if (!IsUsingEF)
if (dbTransaction == null) throw new ArgumentNullException(nameof(dbTransaction)); throw new InvalidOperationException("If you are using the EntityFramework, you need to configure the DbContextType first." +
" otherwise you need to use overloaded method with IDbConnection and IDbTransaction.");
}
return PublishWithTrans(name, content, dbConnection, dbTransaction); private void CheckIsAdoNet(string name)
{
if (name == null) throw new ArgumentNullException(nameof(name));
if (IsUsingEF)
throw new InvalidOperationException("If you are using the EntityFramework, you do not need to use this overloaded.");
} }
private async Task Publish(string name, string content) private async Task PublishCoreAsync(string name, string content)
{ {
var connection = _dbContext.Database.GetDbConnection(); var connection = _dbContext.Database.GetDbConnection();
var transaction = _dbContext.Database.CurrentTransaction; var transaction = _dbContext.Database.CurrentTransaction;
IsCapOpenedTrans = transaction == null; IsCapOpenedTrans = transaction == null;
transaction = transaction ?? await _dbContext.Database.BeginTransactionAsync(IsolationLevel.ReadCommitted); transaction = transaction ?? await _dbContext.Database.BeginTransactionAsync(IsolationLevel.ReadCommitted);
var dbTransaction = transaction.GetDbTransaction(); var dbTransaction = transaction.GetDbTransaction();
await PublishWithTrans(name, content, connection, dbTransaction); await PublishWithTransAsync(name, content, connection, dbTransaction);
}
private void PublishCore(string name, string content)
{
var connection = _dbContext.Database.GetDbConnection();
var transaction = _dbContext.Database.CurrentTransaction;
IsCapOpenedTrans = transaction == null;
transaction = transaction ?? _dbContext.Database.BeginTransaction(IsolationLevel.ReadCommitted);
var dbTransaction = transaction.GetDbTransaction();
PublishWithTrans(name, content, connection, dbTransaction);
}
private async Task PublishWithTransAsync(string name, string content, IDbConnection dbConnection, IDbTransaction dbTransaction)
{
var message = new CapPublishedMessage
{
Name = name,
Content = content,
StatusName = StatusName.Scheduled
};
await dbConnection.ExecuteAsync(PrepareSql(), message, transaction: dbTransaction);
_logger.LogInformation("Message has been persisted in the database. name:" + name);
if (IsCapOpenedTrans)
{
dbTransaction.Commit();
dbTransaction.Dispose();
dbConnection.Dispose();
}
PublishQueuer.PulseEvent.Set();
} }
protected virtual async Task PublishWithTrans(string name, string content, IDbConnection dbConnection, IDbTransaction dbTransaction) private void PublishWithTrans(string name, string content, IDbConnection dbConnection, IDbTransaction dbTransaction)
{ {
var message = new CapPublishedMessage var message = new CapPublishedMessage
{ {
...@@ -91,15 +191,24 @@ namespace DotNetCore.CAP.SqlServer ...@@ -91,15 +191,24 @@ namespace DotNetCore.CAP.SqlServer
Content = content, Content = content,
StatusName = StatusName.Scheduled StatusName = StatusName.Scheduled
}; };
var count = dbConnection.Execute(PrepareSql(), message, transaction: dbTransaction);
_logger.LogInformation("Message has been persisted in the database. name:" + name);
var sql = $"INSERT INTO {_options.Schema}.[Published] ([Name],[Content],[Retries],[Added],[ExpiresAt],[StatusName])VALUES(@Name,@Content,@Retries,@Added,@ExpiresAt,@StatusName)";
await dbConnection.ExecuteAsync(sql, message, transaction: dbTransaction);
if (IsCapOpenedTrans) if (IsCapOpenedTrans)
{ {
dbTransaction.Commit(); dbTransaction.Commit();
dbTransaction.Dispose();
dbConnection.Dispose(); dbConnection.Dispose();
} }
PublishQueuer.PulseEvent.Set(); PublishQueuer.PulseEvent.Set();
} }
private string PrepareSql()
{
return $"INSERT INTO {_options.Schema}.[Published] ([Name],[Content],[Retries],[Added],[ExpiresAt],[StatusName])VALUES(@Name,@Content,@Retries,@Added,@ExpiresAt,@StatusName)";
}
#endregion private methods
} }
} }
\ No newline at end of file
src/DotNetCore.CAP/ICapPublisher.cs
View file @ 5842dfd5
...@@ -9,7 +9,7 @@ namespace DotNetCore.CAP ...@@ -9,7 +9,7 @@ namespace DotNetCore.CAP
public interface ICapPublisher public interface ICapPublisher
{ {
/// <summary> /// <summary>
/// Publish a string message to specified topic. /// (EntityFramework) Asynchronous publish a message.
/// <para> /// <para>
/// If you are using the EntityFramework, you need to configure the DbContextType first. /// If you are using the EntityFramework, you need to configure the DbContextType first.
/// otherwise you need to use overloaded method with IDbConnection and IDbTransaction. /// otherwise you need to use overloaded method with IDbConnection and IDbTransaction.
...@@ -20,7 +20,18 @@ namespace DotNetCore.CAP ...@@ -20,7 +20,18 @@ namespace DotNetCore.CAP
Task PublishAsync(string name, string content); Task PublishAsync(string name, string content);
/// <summary> /// <summary>
/// Publis a object message to specified topic. /// (EntityFramework) Publish a message.
/// <para>
/// If you are using the EntityFramework, you need to configure the DbContextType first.
/// otherwise you need to use overloaded method with IDbConnection and IDbTransaction.
/// </para>
/// </summary>
/// <param name="name">the topic name or exchange router key.</param>
/// <param name="content">message body content.</param>
void Publish(string name, string content);
/// <summary>
/// (EntityFramework) Asynchronous publish a object message.
/// <para> /// <para>
/// If you are using the EntityFramework, you need to configure the DbContextType first. /// If you are using the EntityFramework, you need to configure the DbContextType first.
/// otherwise you need to use overloaded method with IDbConnection and IDbTransaction. /// otherwise you need to use overloaded method with IDbConnection and IDbTransaction.
...@@ -28,24 +39,55 @@ namespace DotNetCore.CAP ...@@ -28,24 +39,55 @@ namespace DotNetCore.CAP
/// </summary> /// </summary>
/// <typeparam name="T">The type of conetent object.</typeparam> /// <typeparam name="T">The type of conetent object.</typeparam>
/// <param name="name">the topic name or exchange router key.</param> /// <param name="name">the topic name or exchange router key.</param>
/// <param name="contentObj">object instance that will be serialized of json.</param> /// <param name="contentObj">message body content, that will be serialized of json.</param>
Task PublishAsync<T>(string name, T contentObj); Task PublishAsync<T>(string name, T contentObj);
/// <summary> /// <summary>
/// Publish a string message to specified topic with transacton. /// (EntityFramework) Publish a object message.
/// <para>
/// If you are using the EntityFramework, you need to configure the DbContextType first.
/// otherwise you need to use overloaded method with IDbConnection and IDbTransaction.
/// </para>
/// </summary> /// </summary>
/// <typeparam name="T">The type of conetent object.</typeparam>
/// <param name="name">the topic name or exchange router key.</param> /// <param name="name">the topic name or exchange router key.</param>
/// <param name="content">message body content.</param> /// <param name="contentObj">message body content, that will be serialized of json.</param>
/// <param name="dbConnection">the dbConnection of <see cref="IDbConnection"/></param> void Publish<T>(string name, T contentObj);
Task PublishAsync(string name, string content, IDbConnection dbConnection);
/// <summary> /// <summary>
/// Publish a string message to specified topic with transacton. /// (ado.net) Asynchronous publish a message.
/// </summary>
/// <param name="name">the topic name or exchange router key.</param>
/// <param name="content">message body content</param>
/// <param name="dbConnection">the connection of <see cref="IDbConnection"/></param>
/// <param name="dbTransaction">the transaction of <see cref="IDbTransaction"/></param>
Task PublishAsync(string name, string content, IDbConnection dbConnection, IDbTransaction dbTransaction = null);
/// <summary>
/// (ado.net) Publish a message.
/// </summary> /// </summary>
/// <param name="name">the topic name or exchange router key.</param> /// <param name="name">the topic name or exchange router key.</param>
/// <param name="content">message body content.</param> /// <param name="content">message body content.</param>
/// <param name="dbConnection">the connection of <see cref="IDbConnection"/></param> /// <param name="dbConnection">the connection of <see cref="IDbConnection"/></param>
/// <param name="dbTransaction">the transaction of <see cref="IDbTransaction"/></param> /// <param name="dbTransaction">the transaction of <see cref="IDbTransaction"/></param>
Task PublishAsync(string name, string content, IDbConnection dbConnection, IDbTransaction dbTransaction); void Publish(string name, string content, IDbConnection dbConnection, IDbTransaction dbTransaction = null);
/// <summary>
/// (ado.net) Asynchronous publish a object message.
/// </summary>
/// <param name="name">the topic name or exchange router key.</param>
/// <param name="contentObj">message body content, that will be serialized of json.</param>
/// <param name="dbConnection">the connection of <see cref="IDbConnection"/></param>
/// <param name="dbTransaction">the transaction of <see cref="IDbTransaction"/></param>
Task PublishAsync<T>(string name, T contentObj, IDbConnection dbConnection, IDbTransaction dbTransaction = null);
/// <summary>
/// (ado.net) Publish a object message.
/// </summary>
/// <param name="name">the topic name or exchange router key.</param>
/// <param name="contentObj">message body content, that will be serialized of json.</param>
/// <param name="dbConnection">the connection of <see cref="IDbConnection"/></param>
/// <param name="dbTransaction">the transaction of <see cref="IDbTransaction"/></param>
void Publish<T>(string name, T contentObj, IDbConnection dbConnection, IDbTransaction dbTransaction = null);
} }
} }
\ No newline at end of file
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