diff --git a/Admin.NET/Admin.NET.Core/Admin.NET.Core.csproj b/Admin.NET/Admin.NET.Core/Admin.NET.Core.csproj
index 7fe0210b..0930f88d 100644
--- a/Admin.NET/Admin.NET.Core/Admin.NET.Core.csproj
+++ b/Admin.NET/Admin.NET.Core/Admin.NET.Core.csproj
@@ -38,7 +38,7 @@
-
+
diff --git a/Admin.NET/Admin.NET.Core/Attribute/DictAttribute.cs b/Admin.NET/Admin.NET.Core/Attribute/DictAttribute.cs
index d252cf03..84e52f64 100644
--- a/Admin.NET/Admin.NET.Core/Attribute/DictAttribute.cs
+++ b/Admin.NET/Admin.NET.Core/Attribute/DictAttribute.cs
@@ -40,7 +40,7 @@ public class DictAttribute : ValidationAttribute, ITransient
// 是否忽略空字符串
if (AllowEmptyStrings && string.IsNullOrEmpty(valueAsString)) return ValidationResult.Success;
- var sysDictDataServiceProvider = validationContext.GetRequiredService();
+ var sysDictDataServiceProvider = App.GetRequiredService();
var dictDataList = sysDictDataServiceProvider.GetDataList(DictTypeCode).Result;
// 使用HashSet来提高查找效率
diff --git a/Admin.NET/Admin.NET.Core/EventBus/RedisEventSourceStorer.cs b/Admin.NET/Admin.NET.Core/EventBus/RedisEventSourceStorer.cs
index 475bdf41..f507e7ba 100644
--- a/Admin.NET/Admin.NET.Core/EventBus/RedisEventSourceStorer.cs
+++ b/Admin.NET/Admin.NET.Core/EventBus/RedisEventSourceStorer.cs
@@ -1,4 +1,4 @@
-// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
+// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
//
// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
//
@@ -35,10 +35,7 @@ public sealed class RedisEventSourceStorer : IEventSourceStorer, IDisposable
private RedisStream _queueBroadcast;
- ///
- /// 路由键
- ///
- private readonly string _routeKey;
+ private ILogger _logger;
///
/// 构造函数
@@ -48,6 +45,8 @@ public sealed class RedisEventSourceStorer : IEventSourceStorer, IDisposable
/// 存储器最多能够处理多少消息，超过该容量进入等待写入
public RedisEventSourceStorer(ICacheProvider cacheProvider, string routeKey, int capacity)
{
+ _logger = App.GetRequiredService>();
+
// 配置通道，设置超出默认容量后进入等待
var boundedChannelOptions = new BoundedChannelOptions(capacity)
{
@@ -58,7 +57,6 @@ public sealed class RedisEventSourceStorer : IEventSourceStorer, IDisposable
_channel = Channel.CreateBounded(boundedChannelOptions);
//_redis = redis as FullRedis;
- _routeKey = routeKey;
// 创建广播消息订阅者，即所有服务器节点都能收到消息（用来发布重启、Reload配置等消息）
FullRedis redis = (FullRedis)cacheProvider.Cache;
@@ -73,23 +71,29 @@ public sealed class RedisEventSourceStorer : IEventSourceStorer, IDisposable
_eventConsumer = new EventConsumer(_queueSingle);
// 订阅消息写入 Channel
- _eventConsumer.Received += (send, cr) =>
+ _eventConsumer.Received += async (send, cr) =>
{
- var oriColor = Console.ForegroundColor;
- ChannelEventSource ces = (ChannelEventSource)cr;
- ConsumeChannelEventSource(ces);
+ // var oriColor = Console.ForegroundColor;
+ try
+ {
+ ChannelEventSource ces = (ChannelEventSource)cr;
+ await ConsumeChannelEventSourceAsync(ces, ces.CancellationToken);
+ }
+ catch (Exception e)
+ {
+ _logger.LogError(e, "处理Received中的消息产生错误！");
+ }
};
_eventConsumer.Start();
}
- private Task OnConsumeBroadcast(string source, Message message, CancellationToken token)
+ private async Task OnConsumeBroadcast(string source, Message message, CancellationToken token)
{
ChannelEventSource ces = JsonConvert.DeserializeObject(source);
- ConsumeChannelEventSource(ces);
- return Task.CompletedTask;
+ await ConsumeChannelEventSourceAsync(ces, token);
}
- private void ConsumeChannelEventSource(ChannelEventSource ces)
+ private async Task ConsumeChannelEventSourceAsync(ChannelEventSource ces, CancellationToken cancel = default)
{
// 打印测试事件
if (ces.EventId != null && ces.EventId.IndexOf(":Test") > 0)
@@ -99,7 +103,7 @@ public sealed class RedisEventSourceStorer : IEventSourceStorer, IDisposable
Console.WriteLine($"有消息要处理{ces.EventId},{ces.Payload}");
Console.ForegroundColor = oriColor;
}
- _channel.Writer.WriteAsync(ces);
+ await _channel.Writer.WriteAsync(ces, cancel);
}
///
diff --git a/Admin.NET/Admin.NET.Core/Service/Wechat/SysWechatPayService.cs b/Admin.NET/Admin.NET.Core/Service/Wechat/SysWechatPayService.cs
index 4220ac1d..04dd1969 100644
--- a/Admin.NET/Admin.NET.Core/Service/Wechat/SysWechatPayService.cs
+++ b/Admin.NET/Admin.NET.Core/Service/Wechat/SysWechatPayService.cs
@@ -19,10 +19,10 @@ public class SysWechatPayService : IDynamicApiController, ITransient
private readonly WechatTenpayClient _wechatTenpayClient;
- public SysWechatPayService(SqlSugarRepository sysWechatPayRep
- , SqlSugarRepository sysWechatRefundRep
- , IOptions wechatPayOptions
- , IOptions payCallBackOptions)
+ public SysWechatPayService(SqlSugarRepository sysWechatPayRep,
+ SqlSugarRepository sysWechatRefundRep,
+ IOptions wechatPayOptions,
+ IOptions payCallBackOptions)
{
_sysWechatPayRep = sysWechatPayRep;
this._sysWechatRefundRep = sysWechatRefundRep;
@@ -378,7 +378,7 @@ public class SysWechatPayService : IDynamicApiController, ITransient
wechatPay.TradeType = response.TradeType; // 交易类型
wechatPay.TradeState = response.TradeState; // 交易状态
wechatPay.TradeStateDescription = response.TradeStateDescription; // 交易状态描述
- wechatPay.BankType = response.BankType; // 付款银行类型
+ wechatPay.BankType = response.BankType; // 付款银行类型
wechatPay.PayerTotal = response.Amount?.PayerTotal; // 用户支付金额
wechatPay.SuccessTime = response.SuccessTime; // 支付完成时间
await _sysWechatPayRep.AsUpdateable(wechatPay).IgnoreColumns(true).ExecuteCommandAsync();
@@ -420,7 +420,7 @@ public class SysWechatPayService : IDynamicApiController, ITransient
wechatPay.TradeType = response.TradeType; // 交易类型
wechatPay.TradeState = response.TradeState; // 交易状态
wechatPay.TradeStateDescription = response.TradeStateDescription; // 交易状态描述
- wechatPay.BankType = response.BankType; // 付款银行类型
+ wechatPay.BankType = response.BankType; // 付款银行类型
wechatPay.PayerTotal = response.Amount?.PayerTotal; // 用户支付金额
wechatPay.SuccessTime = response.SuccessTime; // 支付完成时间
await _sysWechatPayRep.AsUpdateable(wechatPay).IgnoreColumns(true).ExecuteCommandAsync();