admin/JiShe.IOT.Admin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: admin:28fad6b79b411bcf9bf9991b25e28d0206f676db
Branches Tags
admin:master
admin:master_net9_Pomelo
..
compare: admin:b3d7e3d9025b2adafae5b83f2864e9f08d1ad8ff
Branches Tags
admin:master
admin:master_net9_Pomelo

No commits in common. "28fad6b79b411bcf9bf9991b25e28d0206f676db" and "b3d7e3d9025b2adafae5b83f2864e9f08d1ad8ff" have entirely different histories.
28fad6b79b ... b3d7e3d902

5 changed files with 36 additions and 75 deletions
Show Stats Expand all files Collapse all files

2
JiShe.ServicePro

@ -1 +1 @@
Subproject commit 2cc68d62afbcde8e8cc7bfdc57c25ecf6fcbc54e
Subproject commit 3dbd0ebf5fff82ae03945631aa8e401620931ab1

13
host/JiShe.IoT.HttpApi.Host/appsettings.Development.json
View File

@ -32,11 +32,10 @@
"TaskThreadCount": -1
},
"Pulsar": {
"ServiceUrl": "pulsar://47.110.60.222:9093",
"WebUrl": "http://47.110.60.222:9094",
"UserName": "admin",
"ServiceUrl": "pulsar://192.168.56.101:6650",
"WebUrl": "http://192.168.56.101:8080",
"TenantName": "1YMVZZkAkRArjxSD8457",
"Namespace": "OneNET",
"Namespace": "default",
"PulsarSecretKey": "0fd7afb8b0d04e6abc4fdfdac2190a79",
"PulsarSubscriptionCustomName": "sub",
"EnableTls": false,
@ -59,10 +58,8 @@
"IoTDBOptions": {
"UserName": "root",
"Password": "Lixiao@1980",
//"TreeModelClusterList": [ "47.110.53.196:6667", "47.110.60.222:6667", "47.110.62.104:6667" ],
//"TableModelClusterList": [ "47.110.53.196:6667", "47.110.60.222:6667", "47.110.62.104:6667" ],
"TreeModelClusterList": [ "47.110.53.196:30710" ],
"TableModelClusterList": [ "47.110.53.196:30710" ],
"TreeModelClusterList": [ "47.110.53.196:6667", "47.110.60.222:6667", "47.110.62.104:6667" ],
"TableModelClusterList": [ "47.110.53.196:6667", "47.110.60.222:6667", "47.110.62.104:6667" ],
"PoolSize": 32,
"DataBaseName": "jisheiotdata",
"OpenDebugMode": true,

10
src/JiShe.IoT.Application.Contracts/Subscribers/IServiceCommunicationChannelSubscriberService.cs
View File

@ -1,4 +1,7 @@
namespace JiShe.ServicePro.OneNETManagement.Subscribers
using JiShe.ServicePro.Dto;
using JiShe.ServicePro.Kafka.Internal;
namespace JiShe.ServicePro.OneNETManagement.Subscribers
{
/// <summary>
/// 数据通讯通道订阅者服务
@ -6,9 +9,10 @@
public interface IServiceCommunicationChannelSubscriberService
{
/// <summary>
/// 设备上下线通知 Redis 消息订阅
/// 设备上下线通知Kafka消息订阅
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
Task ServiceCommunicationDeviceStatusSubscriber();
Task<ISubscribeAck> ServiceCommunicationDeviceStatusSubscriber(DeviceStatusMessage message);
}
}

4
src/JiShe.IoT.Application/IoTApplicationModule.cs
View File

@ -8,7 +8,6 @@ using JiShe.ServicePro.FreeRedisProvider;
using JiShe.ServicePro.FreeSqlProvider;
using JiShe.ServicePro.IoTDBManagement;
using JiShe.ServicePro.OneNETManagement;
using JiShe.ServicePro.OneNETManagement.Subscribers;
using JiShe.ServicePro.TemplateManagement;
using Microsoft.Extensions.DependencyInjection;
using Volo.Abp;
@ -59,9 +58,6 @@ namespace JiShe.IoT
{
var commonService = context.ServiceProvider.GetRequiredService<ICommonService>();
commonService.InitSelectTypetList();
var communicationChannelService = context.ServiceProvider.GetRequiredService<IServiceCommunicationChannelSubscriberService>();
communicationChannelService.ServiceCommunicationDeviceStatusSubscriber();
}
}
}

76
src/JiShe.IoT.Application/Subscribers/ServiceCommunicationChannelSubscriberService.cs
View File

@ -1,6 +1,4 @@
using FreeRedis;
using FreeSql;
using JiShe.IoT;
using JiShe.IoT;
using JiShe.ServicePro.ApacheIoTDB.Provider.Model;
using JiShe.ServicePro.ApacheIoTDB.Provider.Options;
using JiShe.ServicePro.Consts;
@ -10,7 +8,6 @@ using JiShe.ServicePro.DeviceManagement.DeviceInfos;
using JiShe.ServicePro.Dto;
using JiShe.ServicePro.Enums;
using JiShe.ServicePro.FreeRedisProvider;
using JiShe.ServicePro.FreeSqlProvider;
using JiShe.ServicePro.IoTDBManagement.DataChannels;
using JiShe.ServicePro.IoTDBManagement.TableModels;
using JiShe.ServicePro.Kafka.Attributes;
@ -28,85 +25,51 @@ using Volo.Abp.Auditing;
namespace JiShe.ServicePro.OneNETManagement.Subscribers
{
/// <summary>
/// 数据通讯通道订阅者服务
/// 设备状态统一订阅订阅服务
/// </summary>
[DisableAuditing]
public class ServiceCommunicationChannelSubscriberService : IoTAppService, IServiceCommunicationChannelSubscriberService
public class ServiceCommunicationChannelSubscriberService : IoTAppService, IServiceCommunicationChannelSubscriberService, IKafkaSubscribe
{
private readonly ILogger<ServiceCommunicationChannelSubscriberService> _logger;
private readonly IoTDBOptions _iotDBOptions;
private readonly IRedisPubSubService _redisPubSubService;
/// <summary>
/// 设备状态统一订阅订阅服务
/// </summary>
/// <param name="logger"></param>
/// <param name="iotDBOptions"></param>
/// <param name="redisPubSubService"></param>
public ServiceCommunicationChannelSubscriberService(ILogger<ServiceCommunicationChannelSubscriberService> logger, IOptions<IoTDBOptions> iotDBOptions, IRedisPubSubService redisPubSubService)
public ServiceCommunicationChannelSubscriberService(ILogger<ServiceCommunicationChannelSubscriberService> logger, IOptions<IoTDBOptions> iotDBOptions)
{
_logger = logger;
_iotDBOptions = iotDBOptions.Value;
_redisPubSubService = redisPubSubService;
}
/// <summary>
/// 设备上下线通知 Redis 消息订阅
/// 设备上下线通知Kafka消息订阅
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
public async Task ServiceCommunicationDeviceStatusSubscriber()
[KafkaSubscribe(KafkaTopicConsts.ServiceCommunicationDeviceStatusEventName)]
public async Task<ISubscribeAck> ServiceCommunicationDeviceStatusSubscriber(DeviceStatusMessage message)
{
try
{
// 为订阅回调创建独立的 FreeSql 客户端
var callbackFreeSqlDbContext = FreeSqlDbContext;
// 订阅频道
await _redisPubSubService.SubscribeAsync<DeviceStatusMessage>(RedisConst.ServiceCommunicationDeviceStatusEventName, (message) =>
{
_logger.LogWarning($"Redis订阅收到设备状态消息: {message}");
HandDeviceStatus(message, callbackFreeSqlDbContext).ConfigureAwait(false).GetAwaiter().GetResult();
});
}
catch (Exception)
{
throw;
}
}
/// <summary>
/// 设备状态处理
/// </summary>
/// <param name="deviceStatusMessage"></param>
/// <param name="callbackFreeSqlDbContext"></param>
/// <returns></returns>
/// <exception cref="UserFriendlyException"></exception>
private async Task HandDeviceStatus(DeviceStatusMessage deviceStatusMessage, IFreeSqlProvider callbackFreeSqlDbContext)
{
try
{
if (deviceStatusMessage == null || callbackFreeSqlDbContext == null)
{
throw new UserFriendlyException($"设备状态消息处理失败，参数为空");
}
var deviceEntity = await callbackFreeSqlDbContext.Instance.Select<DeviceManagementInfo>()
.Where(d => d.IoTPlatform == deviceStatusMessage.IoTPlatform && d.DeviceAddress == deviceStatusMessage.DeviceAddress)
var deviceEntity = await FreeSqlDbContext.Instance.Select<DeviceManagementInfo>()
.Where(d => d.IoTPlatform == message.IoTPlatform && d.DeviceAddress == message.DeviceAddress)
.FirstAsync();
if (deviceEntity == null)
{
throw new UserFriendlyException($"{deviceStatusMessage.DeviceAddress} 设备不存在");
return SubscribeAck.Fail();
}
deviceEntity.DeviceOnlineStatus = deviceStatusMessage.Status;
if (deviceStatusMessage.Status == DeviceOnlineStatusEnum.Online)
deviceEntity.DeviceOnlineStatus = message.Status;
if (message.Status == DeviceOnlineStatusEnum.Online)
{
deviceEntity.LastOnlineTime = TimestampHelper.ConvertToDateTime(deviceStatusMessage.ReceivedTime, TimestampUnit.Milliseconds, DateTimeKind.Local);
deviceEntity.LastOnlineTime = TimestampHelper.ConvertToDateTime(message.ReceivedTime, TimestampUnit.Milliseconds,DateTimeKind.Local);
}
else if (deviceStatusMessage.Status == DeviceOnlineStatusEnum.Offline)
else if (message.Status == DeviceOnlineStatusEnum.Offline)
{
deviceEntity.LastOfflineTime = TimestampHelper.ConvertToDateTime(deviceStatusMessage.ReceivedTime, TimestampUnit.Milliseconds, DateTimeKind.Local);
deviceEntity.LastOfflineTime = TimestampHelper.ConvertToDateTime(message.ReceivedTime, TimestampUnit.Milliseconds, DateTimeKind.Local);
}
var updateResult = await FreeSqlDbContext.Instance.Update<DeviceManagementInfo>()
@ -116,9 +79,8 @@ namespace JiShe.ServicePro.OneNETManagement.Subscribers
if (updateResult <=0)
{
_logger.LogError($"{nameof(ServiceCommunicationDeviceStatusSubscriber)} 设备状态更新失败{deviceStatusMessage.Serialize()}");
throw new UserFriendlyException($"设备状态更新失败");
_logger.LogError($"{nameof(ServiceCommunicationDeviceStatusSubscriber)} 设备状态更新失败{message.Serialize()}");
return SubscribeAck.Fail();
}
//更新设备数据缓存到Redis
@ -127,6 +89,8 @@ namespace JiShe.ServicePro.OneNETManagement.Subscribers
deviceCacheInfos.PlatformPassword = null;
RedisProvider.Instance.HSet<DeviceCacheInfos>(RedisConst.CacheAllDeviceInfoHashKey, deviceEntity.DeviceAddress, deviceCacheInfos);
return SubscribeAck.Success();
}
catch (Exception)
{