diff --git a/src/DotNetCore.CAP.MySql/CapPublisher.cs b/src/DotNetCore.CAP.MySql/CapPublisher.cs index 59801df..aac528f 100644 --- a/src/DotNetCore.CAP.MySql/CapPublisher.cs +++ b/src/DotNetCore.CAP.MySql/CapPublisher.cs @@ -7,10 +7,11 @@ using DotNetCore.CAP.Models; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Storage; using Microsoft.Extensions.Logging; +using MySql.Data.MySqlClient; namespace DotNetCore.CAP.MySql { - public class CapPublisher : CapPublisherBase + public class CapPublisher : CapPublisherBase, ICallbackPublisher { private readonly ILogger _logger; private readonly MySqlOptions _options; @@ -61,6 +62,13 @@ namespace DotNetCore.CAP.MySql _logger.LogInformation("Published Message has been persisted in the database. name:" + message.ToString()); } + public async Task PublishAsync(CapPublishedMessage message) + { + using (var conn = new MySqlConnection(_options.ConnectionString)) + { + await conn.ExecuteAsync(PrepareSql(), message); + } + } #region private methods private string PrepareSql() @@ -68,8 +76,6 @@ namespace DotNetCore.CAP.MySql return $"INSERT INTO `{_options.TableNamePrefix}.published` (`Name`,`Content`,`Retries`,`Added`,`ExpiresAt`,`StatusName`)VALUES(@Name,@Content,@Retries,@Added,@ExpiresAt,@StatusName)"; } - - #endregion private methods } } \ No newline at end of file diff --git a/src/DotNetCore.CAP.PostgreSql/CapPublisher.cs b/src/DotNetCore.CAP.PostgreSql/CapPublisher.cs index 4e3c9cd..5a81931 100644 --- a/src/DotNetCore.CAP.PostgreSql/CapPublisher.cs +++ b/src/DotNetCore.CAP.PostgreSql/CapPublisher.cs @@ -7,10 +7,11 @@ using DotNetCore.CAP.Abstractions; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Storage; using Microsoft.Extensions.Logging; +using Npgsql; namespace DotNetCore.CAP.PostgreSql { - public class CapPublisher : CapPublisherBase + public class CapPublisher : CapPublisherBase, ICallbackPublisher { private readonly ILogger _logger; private readonly PostgreSqlOptions _options; @@ -61,9 +62,21 @@ namespace DotNetCore.CAP.PostgreSql _logger.LogInformation("Published Message has been persisted in the database. name:" + message.ToString()); } + public async Task PublishAsync(CapPublishedMessage message) + { + using (var conn = new NpgsqlConnection(_options.ConnectionString)) + { + await conn.ExecuteAsync(PrepareSql(), message); + } + } + + #region private methods + private string PrepareSql() { return $"INSERT INTO \"{_options.Schema}\".\"published\" (\"Name\",\"Content\",\"Retries\",\"Added\",\"ExpiresAt\",\"StatusName\")VALUES(@Name,@Content,@Retries,@Added,@ExpiresAt,@StatusName)"; - } + } + + #endregion } } \ No newline at end of file