diff --git a/src/JiShe.IoT.Application.Contracts/BusinessSystemAggregation/Dto/BatchQueryDeviceInfoInput.cs b/src/JiShe.IoT.Application.Contracts/BusinessSystemAggregation/Dto/BatchQueryDeviceInfoInput.cs new file mode 100644 index 0000000..1f1cbf2 --- /dev/null +++ b/src/JiShe.IoT.Application.Contracts/BusinessSystemAggregation/Dto/BatchQueryDeviceInfoInput.cs @@ -0,0 +1,40 @@ +using JiShe.ServicePro.Enums; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace JiShe.IoT.BusinessSystemAggregation.Dto +{ + /// + /// 批量查询设备信息输入 + /// + public class BatchQueryDeviceInfoInput + { + /// + /// 设备类型 + /// + public DeviceTypeEnum DeviceType { get; set; } + + /// + /// 数据类型 + /// + public string IoTDataType { get; set; } + + /// + /// 开始时间，最终需要转换为纳秒级时间戳 + /// + public DateTime BeginTime { get; set; } + + /// + /// 结束时间，最终需要转换为纳秒级时间戳 + /// + public DateTime EndTime { get; set; } + + /// + /// 设备地址集合 + /// + public List DeviceAddresses { get; set; } + } +} diff --git a/src/JiShe.IoT.Application.Contracts/BusinessSystemAggregation/IBusinessSystemAggregationService.cs b/src/JiShe.IoT.Application.Contracts/BusinessSystemAggregation/IBusinessSystemAggregationService.cs index 8105f1d..cd225e0 100644 --- a/src/JiShe.IoT.Application.Contracts/BusinessSystemAggregation/IBusinessSystemAggregationService.cs +++ b/src/JiShe.IoT.Application.Contracts/BusinessSystemAggregation/IBusinessSystemAggregationService.cs @@ -1,4 +1,5 @@ using JiShe.ServicePro; +using JiShe.ServicePro.ApacheIoTDB.Provider.Model; using System; using System.Collections.Generic; using System.Linq; @@ -13,10 +14,17 @@ namespace JiShe.IoT.BusinessSystemAggregation public interface IBusinessSystemAggregationService : IApplicationService { /// - /// 接收业务系统指令信息 + /// 接收业务系统指令信息，Msg 字段为 ReceiveCommandInfoDto 实体 /// /// /// Task ReceiveCommandInfoAsync(OpenApiRequest input); + + /// + /// 业务系统批量查询设备数据，Msg 字段为 BatchQueryDeviceInfoInput 实体 + /// + /// + /// + Task>> BatchQueryDeviceInfoAsync(OpenApiRequest input); } } diff --git a/src/JiShe.IoT.Application.Contracts/JiShe.IoT.Application.Contracts.csproj b/src/JiShe.IoT.Application.Contracts/JiShe.IoT.Application.Contracts.csproj index 527e95e..b240040 100644 --- a/src/JiShe.IoT.Application.Contracts/JiShe.IoT.Application.Contracts.csproj +++ b/src/JiShe.IoT.Application.Contracts/JiShe.IoT.Application.Contracts.csproj @@ -43,8 +43,4 @@ - - - - \ No newline at end of file diff --git a/src/JiShe.IoT.Application/BusinessSystemAggregation/BusinessSystemAggregationService.cs b/src/JiShe.IoT.Application/BusinessSystemAggregation/BusinessSystemAggregationService.cs index 7ee6879..506f189 100644 --- a/src/JiShe.IoT.Application/BusinessSystemAggregation/BusinessSystemAggregationService.cs +++ b/src/JiShe.IoT.Application/BusinessSystemAggregation/BusinessSystemAggregationService.cs @@ -1,4 +1,6 @@ -using JiShe.ServicePro; +using JiShe.IoT.BusinessSystemAggregation.Dto; +using JiShe.ServicePro; +using JiShe.ServicePro.ApacheIoTDB.Provider.Model; using JiShe.ServicePro.ApacheIoTDB.Provider.Options; using JiShe.ServicePro.Core; using JiShe.ServicePro.DataChannelManages; @@ -8,21 +10,24 @@ using JiShe.ServicePro.Encrypt; using JiShe.ServicePro.Enums; using JiShe.ServicePro.FreeRedisProvider; using JiShe.ServicePro.IoTDBManagement.DataChannels; +using JiShe.ServicePro.IoTDBManagement.TreeModels; using Mapster; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; +using System.Collections.Generic; namespace JiShe.IoT.BusinessSystemAggregation { /// /// 业务系统聚合服务 /// - public class BusinessSystemAggregationService(IOptions options, IReliableRedisPubSubService redisPubSubService, IDeviceAppService deviceAppService, IIoTDBDataChannelManageService ioTDBDataChannelManageService, IOptions _ioTDBOptions) : IoTAppService, IBusinessSystemAggregationService + public class BusinessSystemAggregationService(IOptions options, IReliableRedisPubSubService redisPubSubService, IDeviceAppService deviceAppService, IIoTDBDataChannelManageService ioTDBDataChannelManageService, IOptions _ioTDBOptions, ITreeModelService treeModelService, ILogger _logger) : IoTAppService, IBusinessSystemAggregationService { ServerApplicationOptions srverOptions = options.Value; IoTDBOptions ioTDBOptions = _ioTDBOptions.Value; /// - /// 接收业务系统指令信息，缓存进Kafka + /// 接收业务系统指令信息 /// [AllowAnonymous] public async Task ReceiveCommandInfoAsync(OpenApiRequest input) @@ -30,35 +35,44 @@ namespace JiShe.IoT.BusinessSystemAggregation try { - bool verifySignatureReult = EncryptUtil.OpenApiVerifySignature(input.Message, input.Nonce, input.Timestamp, input.Signature, srverOptions.SignatureToken); - if (verifySignatureReult == false)//签名校验失败 - { - return HttpDataResultExtensions.Failed("签名校验失败", -101, ResponeResultEnum.NotAllowed); - } - if (string.IsNullOrWhiteSpace(input.Message)) + var handleResult = HandleOpenApiRequest(input); + if (handleResult.Success == false) { - return HttpDataResultExtensions.Failed("指令下发内容不能为空", -102, ResponeResultEnum.Fail); + return handleResult; } - - //判断是否需要解密 - ReceiveCommandInfoDto messageBody = null; + var messageBody = handleResult.Data; string tempMessageBody = null; - if (srverOptions.IsAesEncrypted && !string.IsNullOrWhiteSpace(srverOptions.AesSecurityKey)) - { - tempMessageBody = EncryptUtil.OpenApiDecrypto(input.Message, srverOptions.AesSecurityKey); - messageBody = tempMessageBody.Deserialize(); - } - else - { - messageBody = input.Message.Deserialize(); - tempMessageBody = input.Message; - } - if (messageBody == null) - { - return HttpDataResultExtensions.Failed("指令下发内容不能为空", -103, ResponeResultEnum.Fail); - } + //bool verifySignatureReult = EncryptUtil.OpenApiVerifySignature(input.Message, input.Nonce, input.Timestamp, input.Signature, srverOptions.SignatureToken); + //if (verifySignatureReult == false)//签名校验失败 + //{ + // return HttpDataResultExtensions.Failed("签名校验失败", -101, ResponeResultEnum.NotAllowed); + //} + + //if (string.IsNullOrWhiteSpace(input.Message)) + //{ + // return HttpDataResultExtensions.Failed("指令下发内容不能为空", -102, ResponeResultEnum.Fail); + //} + + ////判断是否需要解密 + //ReceiveCommandInfoDto messageBody = null; + //string tempMessageBody = null; + //if (srverOptions.IsAesEncrypted && !string.IsNullOrWhiteSpace(srverOptions.AesSecurityKey)) + //{ + // tempMessageBody = EncryptUtil.OpenApiDecrypto(input.Message, srverOptions.AesSecurityKey); + // messageBody = tempMessageBody.Deserialize(); + //} + //else + //{ + // tempMessageBody = input.Message; + // messageBody = input.Message.Deserialize(); + //} + + //if (messageBody == null) + //{ + // return HttpDataResultExtensions.Failed("指令下发内容不能为空", -103, ResponeResultEnum.Fail); + //} //限定来源类型必须为业务系统 if (messageBody.SourceType != DeviceTelemetrySourceTypeEnum.BusinessSystem) @@ -100,5 +114,140 @@ namespace JiShe.IoT.BusinessSystemAggregation return HttpDataResultExtensions.Failed($"指令处理失败，发送异常：{ex.Message}", -106); } } + + + /// + /// 业务系统批量查询设备数据，Msg 字段为BatchQueryDeviceInfoInput实体 + /// + /// + /// + [AllowAnonymous] + public async Task>> BatchQueryDeviceInfoAsync(OpenApiRequest input) + { + try + { + var handleResult = HandleOpenApiRequest(input); + if (handleResult.Success == false) + { + return HttpDataResultExtensions.Failed>(null, handleResult.Message, handleResult.LocationCode); + } + var messageBody = handleResult.Data; + + if (messageBody.DeviceAddresses == null || messageBody.DeviceAddresses.Count <= 0) + { + return HttpDataResultExtensions.Failed>(null, "设备地址不能为空", -103); + } + + // Lua脚本 + string luaScript = @" + local hashKey = KEYS[1] + local fieldKeys = ARGV + return redis.call('HMGET', hashKey, unpack(fieldKeys))"; + + //执行脚本 + var result = await FreeRedisProvider.Instance.EvalAsync + ( + luaScript, + new[] { RedisConst.CacheAllDeviceInfoHashKey }, + messageBody.DeviceAddresses.ToArray() + ); + + List deviceCacheInfos = new List(); + + // 处理返回结果 + if (result is object[] values) + { + foreach (var value in values) + { + var tempFocusInfo = ServiceProJsonSerializer.Deserialize(value as string); + deviceCacheInfos.Add(tempFocusInfo); + } + } + + List queryResult = new List(); + foreach (var deviceAddress in messageBody.DeviceAddresses) + { + var deviceCacheInfo = deviceCacheInfos.FirstOrDefault(x => x.DeviceAddress == deviceAddress); + + if (deviceCacheInfo == null) + { + _logger.LogError($"{nameof(BatchQueryDeviceInfoAsync)} 业务系统批量查询设备数据，设备地址：{deviceAddress}未找到设备地址缓存信息，消息体为：{input.Serialize()}"); + continue; + } + + var pageResult = await treeModelService.DeviceDataInfoPageAsync(new DeviceTreeModelDataInfoInput() + { + DeviceAddress = deviceAddress, + DeviceType = messageBody.DeviceType, + IoTDataType = messageBody.IoTDataType, + IsNeedPaging = false, + }); + + //todo 根据业务系统时间间隔要求进行过滤 + if (pageResult.Items != null && pageResult.Items.Count > 0) + { + queryResult.AddRange(pageResult.Items); + } + } + + return HttpDataResultExtensions.Success(queryResult,"查询成功"); + + + } + catch (Exception ex) + { + + return HttpDataResultExtensions.Failed>(null, $"查询设备数据失败，发送异常：{ex.Message}", -106); + + } + } + + + /// + /// 处理开放接口请求 + /// + /// + /// + /// + private HttpDataResult HandleOpenApiRequest(OpenApiRequest input) where T : class + { + if (string.IsNullOrWhiteSpace(input.Message) || string.IsNullOrWhiteSpace(input.Message) || string.IsNullOrWhiteSpace(input.Signature)) + { + return HttpDataResultExtensions.Failed(null, "请求参数不能为空", -1101); + } + + if (input.Timestamp <= 946656000000)//时间戳小于2000年，视为错误 + { + return HttpDataResultExtensions.Failed(null, "时间戳异常", -1102); + } + + bool verifySignatureReult = EncryptUtil.OpenApiVerifySignature(input.Message, input.Nonce, input.Timestamp, input.Signature, srverOptions.SignatureToken); + if (verifySignatureReult == false)//签名校验失败 + { + return HttpDataResultExtensions.Failed(null, "签名校验失败", -1103); + } + + + //判断是否需要解密 + T messageBody = default; + string tempMessageBody = null; + if (srverOptions.IsAesEncrypted && !string.IsNullOrWhiteSpace(srverOptions.AesSecurityKey)) + { + tempMessageBody = EncryptUtil.OpenApiDecrypto(input.Message, srverOptions.AesSecurityKey); + messageBody = tempMessageBody.Deserialize(); + } + else + { + tempMessageBody = input.Message; + messageBody = input.Message.Deserialize(); + } + + if (messageBody == null) + { + return HttpDataResultExtensions.Failed(null, "获取数据体失败", -1104); + } + + return HttpDataResultExtensions.Success(messageBody, tempMessageBody); + } } } diff --git a/src/JiShe.IoT.Application/DeviceAggregation/DeviceAggregationService.cs b/src/JiShe.IoT.Application/DeviceAggregation/DeviceAggregationService.cs index 1745f38..4b1ca88 100644 --- a/src/JiShe.IoT.Application/DeviceAggregation/DeviceAggregationService.cs +++ b/src/JiShe.IoT.Application/DeviceAggregation/DeviceAggregationService.cs @@ -272,7 +272,7 @@ namespace JiShe.IoT.DeviceAggregation deviceCacheInfos.IoTPlatformResponse = null; deviceCacheInfos.PlatformPassword = null; - RedisProvider.Instance.HSet(RedisConst.CacheAllDeviceInfoHashKey, input.DeviceAddress, deviceCacheInfos); + FreeRedisProvider.Instance.HSet(RedisConst.CacheAllDeviceInfoHashKey, input.DeviceAddress, deviceCacheInfos); diff --git a/src/JiShe.IoT.Application/IoTAppService.cs b/src/JiShe.IoT.Application/IoTAppService.cs index 420d724..ddb0f5e 100644 --- a/src/JiShe.IoT.Application/IoTAppService.cs +++ b/src/JiShe.IoT.Application/IoTAppService.cs @@ -16,7 +16,7 @@ namespace JiShe.IoT public abstract class IoTAppService : ApplicationService { protected IFreeSqlProvider FreeSqlDbContext => LazyServiceProvider.LazyGetRequiredService(); - protected IFreeRedisProvider RedisProvider => LazyServiceProvider.LazyGetRequiredService(); + protected IFreeRedisProvider FreeRedisProvider => LazyServiceProvider.LazyGetRequiredService(); protected IoTAppService() { diff --git a/src/JiShe.IoT.Application/Subscribers/ServiceCommunicationChannelSubscriberService.cs b/src/JiShe.IoT.Application/Subscribers/ServiceCommunicationChannelSubscriberService.cs index edb7ec4..08872d5 100644 --- a/src/JiShe.IoT.Application/Subscribers/ServiceCommunicationChannelSubscriberService.cs +++ b/src/JiShe.IoT.Application/Subscribers/ServiceCommunicationChannelSubscriberService.cs @@ -41,7 +41,7 @@ namespace JiShe.ServicePro.OneNETManagement.Subscribers { // 为订阅回调创建独立的 FreeSql 客户端 var callbackFreeSqlDbContext = FreeSqlDbContext; - var callbackFreeSql = RedisProvider; + var callbackFreeSql = FreeRedisProvider; // 订阅频道