OneNET 数据日志处理重新处理

src/JiShe.IoT.Application.Contracts/IoTPlatformAggregation/IIoTPlatformAggregationService.cs

@@ -1,5 +1,6 @@
 using JiShe.IoT.IoTPlatformAggregation.Dto;
 using JiShe.ServicePro.Commons;
+using JiShe.ServicePro.IoTDBManagement.TableModels;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -45,5 +46,13 @@ namespace JiShe.IoT.IoTPlatformAggregation
         /// <returns></returns>
         Task<object> UpdateIoTPlatformProductThingModelInfoAsync(UpdateIoTPlatformProductThingModelInfoInput input
             );
+
+        /// <summary>
+        /// OneNET 数据日志处理重新处理
+        /// </summary>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        Task IoTPlatformDataLogReHandlerAsync(QueryOneNETReceiveMessageInput input
+            );
     }
 }

src/JiShe.IoT.Application/IoTPlatformAggregation/IoTPlatformAggregationService.cs

@@ -7,8 +7,12 @@ using JiShe.ServicePro.CTWingManagement.CTWingProducts;
 using JiShe.ServicePro.DeviceManagement.ThingModels;
 using JiShe.ServicePro.Enums;
 using JiShe.ServicePro.FreeRedisProvider;
+using JiShe.ServicePro.IoTDBManagement.TableModels;
+using JiShe.ServicePro.IoTDBManagement.TreeModels;
+using JiShe.ServicePro.OneNET.Provider.ReceiveModels;
 using JiShe.ServicePro.OneNETManagement.OneNETAccounts;
 using JiShe.ServicePro.OneNETManagement.OneNETProducts;
+using JiShe.ServicePro.OneNETManagement.Subscribers;
 using Microsoft.AspNetCore.DataProtection.KeyManagement;
 using Microsoft.Extensions.Logging;
 using System;
@@ -31,6 +35,9 @@ namespace JiShe.IoT.IoTPlatformAggregation
         private readonly ICTWingAccountService _ctwingAccountService;
         private readonly IOneNETAccountService _oneNETAccountService;
         private readonly IIoTPlatformThingModelInfoAppService _ioTPlatformThingModelInfoAppService;
+        private readonly IOneNETSubscriberBasic _oneNETSubscriberBasicService;
+        private readonly ITableModelService _tableModelService;
+
 
 
         public IoTPlatformAggregationService(ILogger<IoTPlatformAggregationService> logger,
@@ -38,6 +45,8 @@ namespace JiShe.IoT.IoTPlatformAggregation
             IOneNETProductService oneNetProductService,
             ICTWingAccountService ctwingAccountService,
             IOneNETAccountService oneNETAccountService,
+            IOneNETSubscriberBasic oneNETSubscriberBasicService,
+            ITableModelService tableModelService,
             IIoTPlatformThingModelInfoAppService ioTPlatformThingModelInfoAppService)
         {
             _logger = logger;
@@ -46,6 +55,8 @@ namespace JiShe.IoT.IoTPlatformAggregation
             _ctwingAccountService = ctwingAccountService;
             _oneNETAccountService = oneNETAccountService;
             _ioTPlatformThingModelInfoAppService = ioTPlatformThingModelInfoAppService;
+            _oneNETSubscriberBasicService = oneNETSubscriberBasicService;
+            _tableModelService = tableModelService;
         }
 
         /// <summary>
@@ -349,5 +360,37 @@ namespace JiShe.IoT.IoTPlatformAggregation
                 throw;
             }
         }
+
+
+
+        /// <summary>
+        /// OneNET 数据日志处理重新处理
+        /// </summary>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        public async Task IoTPlatformDataLogReHandlerAsync(QueryOneNETReceiveMessageInput input)
+        {
+            try
+            {
+                var oneNETDataLoglist = await _tableModelService.OneNETLogInfoPageAsync(input);
+                if (oneNETDataLoglist == null || oneNETDataLoglist.Items == null || oneNETDataLoglist.Items.Count <= 0)
+                {
+                    _logger.LogError($"{nameof(IoTPlatformDataLogReHandlerAsync)}OneNET数据日志处理重新处理失败，未获取到数据");
+                    return;
+                }
+
+                foreach (var item in oneNETDataLoglist.Items)
+                {
+                    string topicName = string.Equals(item.MessageType, IoTDBDataTypeConst.Data, StringComparison.OrdinalIgnoreCase) ? DistributedMessageCenterConst.OneNETThingModelPropertyChangeReceivedEventName : DistributedMessageCenterConst.OneNETThingModelEventChangeReceivedEventName;
+                    var tempOneNETReceiveBasicModel = item.RawMessage.Deserialize<OneNETReceiveBasicModel>();
+                    await _oneNETSubscriberBasicService.OneNETReceiveThingModelHandlerAsync(topicName, item.IoTDataType, tempOneNETReceiveBasicModel, true);
+                }
+            }
+            catch (Exception)
+            {
+
+                throw;
+            }
+        }
     }
 }

src/JiShe.IoT.HttpApi/Controllers/IoTPlatformAggregationController.cs

@@ -6,6 +6,7 @@ using JiShe.ServicePro;
 using JiShe.ServicePro.Commons;
 using JiShe.ServicePro.Core;
 using JiShe.ServicePro.DeviceManagement.DeviceInfos;
+using JiShe.ServicePro.IoTDBManagement.TableModels;
 
 namespace JiShe.IoT.Controllers
 {
@@ -69,5 +70,18 @@ namespace JiShe.IoT.Controllers
             return await _iotPlatformAggregationService.UpdateIoTPlatformProductThingModelInfoAsync(input);
         }
 
+
+        /// <summary>
+        /// OneNET数据日志处理重新处理
+        /// </summary>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        [HttpPost(nameof(IoTPlatformDataLogReHandlerAsync))]
+        [SwaggerOperation(summary: "OneNET数据日志处理重新处理", Tags = new[] { "AggregationIoTPlatform" })]
+        public async Task IoTPlatformDataLogReHandlerAsync(QueryOneNETReceiveMessageInput input)
+        {
+            await _iotPlatformAggregationService.IoTPlatformDataLogReHandlerAsync(input);
+        }
+
     }
 }