JiShe.ServicePro

@@ -1 +1 @@
-Subproject commit db713eb791b60bff2b85f12aaf0ceb3fa685b7ee
+Subproject commit d8e8f7ca8e93f8c45cad02cf7e646a0d1d9e9e53

host/JiShe.IoT.HttpApi.Host/IoTHttpApiHostModule.cs

@@ -1,4 +1,4 @@
-using JiShe.ServicePro.Core;
+using JiShe.ServicePro.ServerOptions;
 using JiShe.ServicePro.SwaggerConfigs;
 using System.Reflection;
 using Volo.Abp.BlobStoring.FileSystem;
@@ -51,11 +51,6 @@ namespace JiShe.IoT
             {
                 configuration.GetSection(nameof(ServerApplicationOptions)).Bind(options);
             });
-
-            Configure<DataChannelOptions>(options =>
-            {
-                configuration.GetSection(nameof(DataChannelOptions)).Bind(options);
-            });
         }
 
         public override void OnApplicationInitialization(ApplicationInitializationContext context)

host/JiShe.IoT.HttpApi.Host/JiShe.IoT.HttpApi.Host.csproj

@@ -6,8 +6,7 @@
         <RootNamespace>JiShe.IoT</RootNamespace>
         <PreserveCompilationReferences>true</PreserveCompilationReferences>
         <UserSecretsId>JiShe.IoT-4681b4fd-151f-4221-84a4-929d86723e4c</UserSecretsId>
-		<ServerGarbageCollection>true</ServerGarbageCollection>
+    </PropertyGroup>
-	</PropertyGroup>
 
     <ItemGroup>
         <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" />

src/JiShe.IoT.Application/BusinessSystemAggregation/BusinessSystemAggregationService.cs

@@ -1,11 +1,22 @@
-using JiShe.ServicePro;
+using JiShe.IoT.OneNETAggregation.Dto;
+using JiShe.IoT.Workshops;
+using JiShe.ServicePro;
 using JiShe.ServicePro.Core;
 using JiShe.ServicePro.DeviceManagement.DeviceInfos;
 using JiShe.ServicePro.Dto;
 using JiShe.ServicePro.Encrypt;
 using JiShe.ServicePro.Enums;
 using JiShe.ServicePro.FreeRedisProvider;
+using JiShe.ServicePro.Kafka.Consts;
+using JiShe.ServicePro.Kafka.Producer;
+using JiShe.ServicePro.ServerOptions;
 using Microsoft.Extensions.Options;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using System.Xml.Linq;
 
 namespace JiShe.IoT.BusinessSystemAggregation
 {
@@ -64,11 +75,11 @@ namespace JiShe.IoT.BusinessSystemAggregation
                 //将指令存储Kafka的OneNET主题中
                 if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.OneNET)
                 {
-                    await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.OneNETCommandIssuedEventName, input);
+                    await redisPubSubService.PublishReliableAsync(KafkaTopicConsts.OneNETCommandIssuedEventName, input);
                 }
                 else if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.CTWing)
                 {
-                    await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.CTWingAepCommandIssuedEventName, input);
+                    await redisPubSubService.PublishReliableAsync(KafkaTopicConsts.CTWingAepCommandIssuedEventName, input);
                 }
                 else
                 {

src/JiShe.IoT.Application/CTWingAggregation/CTWingAggregationService.cs

@@ -1,12 +1,19 @@
 using JiShe.IoT.CTWingAggregation.Dto;
 using JiShe.IoT.DeviceAggregation;
 using JiShe.IoT.DeviceAggregation.Dto;
+using JiShe.IoT.OneNETAggregation.Dto;
 using JiShe.IoT.Workshops;
 using JiShe.ServicePro;
 using JiShe.ServicePro.Core;
+using JiShe.ServicePro.CTWingManagement.CTWingDevices;
 using JiShe.ServicePro.CTWingManagement.CTWingProduct;
+using JiShe.ServicePro.DeviceManagement.DeviceInfos;
+using JiShe.ServicePro.DeviceManagement.DeviceInfos.Dto;
 using JiShe.ServicePro.Encrypt;
 using JiShe.ServicePro.Enums;
+using JiShe.ServicePro.FreeRedisProvider;
+using JiShe.ServicePro.ServerOptions;
+using Mapster;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 

src/JiShe.IoT.Application/DeviceAggregation/DeviceAggregationService.cs

@@ -7,6 +7,9 @@ using JiShe.ServicePro.DeviceManagement.Permissions;
 using JiShe.ServicePro.Dto;
 using JiShe.ServicePro.Enums;
 using JiShe.ServicePro.FreeRedisProvider;
+using JiShe.ServicePro.Kafka.Consts;
+using JiShe.ServicePro.Kafka.Producer;
+using JiShe.ServicePro.OneNET.Provider.OpenApiModels.Devices;
 using JiShe.ServicePro.OneNETManagement.OneNETDevices;
 using JiShe.ServicePro.OneNETManagement.OneNETProducts;
 using Mapster;
@@ -297,7 +300,7 @@ namespace JiShe.IoT.DeviceAggregation
                 }
                 else if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.CTWing)
                 {
-                    await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.CTWingAepCommandIssuedEventName,commandRequest);
+                    await redisPubSubService.PublishReliableAsync(KafkaTopicConsts.CTWingAepCommandIssuedEventName,commandRequest);
                     return true;
                 }
                 else
@@ -596,7 +599,7 @@ namespace JiShe.IoT.DeviceAggregation
                     throw new UserFriendlyException("设备不在线");
                 }
 
-                await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.OneNETCommandIssuedEventName,  commandRequest);
+                await redisPubSubService.PublishReliableAsync(KafkaTopicConsts.OneNETCommandIssuedEventName,  commandRequest);
                 return true;
             }
             catch (Exception)

src/JiShe.IoT.Application/IoTApplicationModule.cs

@@ -60,8 +60,8 @@ namespace JiShe.IoT
             var commonService = context.ServiceProvider.GetRequiredService<ICommonService>();
             commonService.InitSelectTypetList();
 
-            var serviceCommunicationChannelSubscriberService = context.ServiceProvider.GetRequiredService<IServiceCommunicationChannelSubscriberService>();
+            var issueSubscriberService = context.ServiceProvider.GetRequiredService<IOneNetIssueSubscriberService>();
-            serviceCommunicationChannelSubscriberService.ServiceCommunicationDeviceStatusSubscriber();
+            issueSubscriberService.IssueCommandRedisSubscriber();
         }
     }
 }

src/JiShe.IoT.Application/OneNETAggregation/OneNETAggregationService.cs

@@ -6,7 +6,9 @@ using JiShe.ServicePro;
 using JiShe.ServicePro.Core;
 using JiShe.ServicePro.Encrypt;
 using JiShe.ServicePro.Enums;
+using JiShe.ServicePro.OneNETManagement.OneNETDevices;
 using JiShe.ServicePro.OneNETManagement.OneNETProducts;
+using JiShe.ServicePro.ServerOptions;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 

src/JiShe.IoT.Application/Subscribers/ServiceCommunicationChannelSubscriberService.cs

@@ -45,10 +45,21 @@ namespace JiShe.ServicePro.OneNETManagement.Subscribers
 
 
                 // 订阅频道
-                await _redisPubSubService.SubscribeReliableAsync<DeviceStatusMessage>(DistributedMessageCenterConst.ServiceCommunicationDeviceStatusEventName, async (message) =>
+                await _redisPubSubService.SubscribeReliableAsync<DeviceStatusMessage>(RedisConst.ServiceCommunicationDeviceStatusEventName, async (message) =>
                 {
-                    await HandDeviceStatus(message, callbackFreeSqlDbContext, callbackFreeSql);
+                    try
-                    return true;
+                    {
+                        //_logger.LogWarning($"Redis订阅收到设备状态消息: {message.Serialize()}");
+
+                        HandDeviceStatus(message, callbackFreeSqlDbContext, callbackFreeSql).ConfigureAwait(false).GetAwaiter().GetResult();
+                        return true;
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogError($"处理Redis订阅设备状态消息发生异常:{ex.Message}，数据写入死信队列，消息体： {message.Serialize()}");
+                        await _redisPubSubService.PublishReliableAsync(RedisConst.ServiceCommunicationDeviceStatusDLQ, message);
+                        return false;
+                    }
                 });
             }
             catch (Exception)