xiaowei/DMS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor:删除MqttServerDto,将使用MqttServerDto的全部转为使用MqttServer

Browse Source
This commit is contained in:
David P.G
2025-10-06 19:32:45 +08:00
parent c85f89db33
commit 94ad940e03
17 changed files with 174 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
DMS.Infrastructure/Services/Mqtt/MqttBackgroundService.cs
View File

@@ -186,34 +186,15 @@ namespace DMS.Infrastructure.Services.Mqtt
_mqttServers.Clear();
// 从数据服务中心获取所有激活的MQTT服务器
var mqttServerDtos = _appDataStorageService.MqttServers.Values.ToList();
var mqttServers = _appDataStorageService.MqttServers.Values.ToList();
foreach (var mqttServerDto in mqttServerDtos)
foreach (var mqttServer in mqttServers)
{
// 将 MqttServerDto 转换为 MqttServerConfig
var mqttServer = new MqttServer
{
Id = mqttServerDto.Id,
ServerName = mqttServerDto.ServerName,
ServerUrl = mqttServerDto.ServerUrl,
Port = mqttServerDto.Port,
Username = mqttServerDto.Username,
Password = mqttServerDto.Password,
IsActive = mqttServerDto.IsActive,
SubscribeTopic = mqttServerDto.SubscribeTopic,
PublishTopic = mqttServerDto.PublishTopic,
ClientId = mqttServerDto.ClientId,
CreatedAt = mqttServerDto.CreatedAt,
ConnectedAt = mqttServerDto.ConnectedAt,
ConnectionDuration = mqttServerDto.ConnectionDuration,
MessageFormat = mqttServerDto.MessageFormat
};
_mqttServers.TryAdd(mqttServer.Id, mqttServer);
_mqttServiceManager.AddMqttServer(mqttServer);
}
_logger.LogInformation($"成功加载 {mqttServerDtos.Count} 个MQTT配置");
_logger.LogInformation($"成功加载 {mqttServers.Count} 个MQTT配置");
return true;
}
catch (Exception ex)

25
DMS.Infrastructure/Services/Mqtt/MqttServiceManager.cs
View File

@@ -1,8 +1,7 @@
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Text;
using AutoMapper;
using DMS.Application.DTOs;
using DMS.Application.Events;
using DMS.Application.Interfaces;
using DMS.Core.Enums;
@@ -23,7 +22,7 @@ namespace DMS.Infrastructure.Services.Mqtt
private readonly IAppDataCenterService _appDataCenterService;
private readonly IMqttServiceFactory _mqttServiceFactory;
private readonly IEventService _eventService;
private readonly IMapper _mapper;
private readonly ConcurrentDictionary<int, MqttDeviceContext> _mqttContexts;
private readonly SemaphoreSlim _semaphore;
private bool _disposed = false;
@@ -33,15 +32,13 @@ namespace DMS.Infrastructure.Services.Mqtt
IDataProcessingService dataProcessingService,
IAppDataCenterService appDataCenterService,
IMqttServiceFactory mqttServiceFactory,
IEventService eventService,
IMapper mapper)
IEventService eventService)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_dataProcessingService = dataProcessingService ?? throw new ArgumentNullException(nameof(dataProcessingService));
_appDataCenterService = appDataCenterService ?? throw new ArgumentNullException(nameof(appDataCenterService));
_mqttServiceFactory = mqttServiceFactory ?? throw new ArgumentNullException(nameof(mqttServiceFactory));
_eventService = eventService ?? throw new ArgumentNullException(nameof(eventService));
_mapper = mapper ?? throw new ArgumentNullException(nameof(mapper));
_mqttContexts = new ConcurrentDictionary<int, MqttDeviceContext>();
_semaphore = new SemaphoreSlim(10, 10); // 默认最大并发连接数为10
@@ -187,8 +184,7 @@ namespace DMS.Infrastructure.Services.Mqtt
context.MqttServerConfig.IsConnect = false;
_logger.LogWarning("MQTT服务器 {ServerName} 连接失败", context.MqttServerConfig.ServerName);
}
//触发MQTT连接状态改变事件
_eventService.RaiseMqttServerChanged(this, new MqttServerChangedEventArgs(ActionChangeType.Updated, _mapper.Map<MqttServerDto>(context.MqttServerConfig), MqttServerPropertyType.IsConnect));
_eventService.RaiseMqttServerChanged(this, new MqttServerChangedEventArgs(ActionChangeType.Updated, context.MqttServerConfig, MqttServerPropertyType.IsConnect));
}
catch (Exception ex)
{
@@ -196,7 +192,7 @@ namespace DMS.Infrastructure.Services.Mqtt
context.MqttServerConfig.ServerName, ex.Message);
context.ReconnectAttempts++;
context.MqttServerConfig.IsConnect = false;
_eventService.RaiseMqttServerChanged(this, new MqttServerChangedEventArgs(ActionChangeType.Updated, _mapper.Map<MqttServerDto>(context.MqttServerConfig), MqttServerPropertyType.IsConnect));
_eventService.RaiseMqttServerChanged(this, new MqttServerChangedEventArgs(ActionChangeType.Updated, context.MqttServerConfig, MqttServerPropertyType.IsConnect));
}
finally
{
@@ -220,7 +216,7 @@ namespace DMS.Infrastructure.Services.Mqtt
// 如果连接状态从连接变为断开，触发事件
context.MqttServerConfig.IsConnect = false;
_eventService.RaiseMqttServerChanged(this, new MqttServerChangedEventArgs(ActionChangeType.Updated, _mapper.Map<MqttServerDto>(context.MqttServerConfig), MqttServerPropertyType.IsConnect));
_eventService.RaiseMqttServerChanged(this, new MqttServerChangedEventArgs(ActionChangeType.Updated, context.MqttServerConfig, MqttServerPropertyType.IsConnect));
}
catch (Exception ex)
{
@@ -366,7 +362,7 @@ namespace DMS.Infrastructure.Services.Mqtt
/// <summary>
/// 处理MQTT服务器添加事件
/// </summary>
private void HandleMqttServerAdded(MqttServerDto mqttServer)
private void HandleMqttServerAdded(MqttServer mqttServer)
{
if (mqttServer == null)
{
@@ -379,11 +375,8 @@ namespace DMS.Infrastructure.Services.Mqtt
_logger.LogInformation("处理MQTT服务器添加事件: {MqttServerId} ({MqttServerName})",
mqttServer.Id, mqttServer.ServerName);
// 将DTO转换为MqttServer实体
var mqttServerEntity = _mapper.Map<MqttServer>(mqttServer);
// 添加服务器到监控列表
AddMqttServer(mqttServerEntity);
AddMqttServer(mqttServer);
// 如果服务器是激活状态，则尝试连接
if (mqttServer.IsActive)
@@ -413,7 +406,7 @@ namespace DMS.Infrastructure.Services.Mqtt
/// <summary>
/// 处理MQTT服务器更新事件
/// </summary>
private async Task HandleMqttServerUpdated(MqttServerDto mqttServer, MqttServerPropertyType propertyType)
private async Task HandleMqttServerUpdated(MqttServer mqttServer, MqttServerPropertyType propertyType)
{
if (mqttServer == null)
{