1

src/YSAI.DAQ/YSAI.Core/data/AddressDetails.cs

@@ -13,7 +13,7 @@ namespace YSAI.Core.data
     /// <summary>
     /// 单个地址的详细信息
     /// </summary>
-    public class AddressDetails : AddressRelay
+    public class AddressDetails
     {
         /// <summary>
         /// 可以理解成唯一标识符（可以存机台号、组名、车间、厂）

src/YSAI.DAQ/YSAI.Core/data/AddressHandler.cs

@@ -15,6 +15,8 @@ using YSAI.Core.reflection;
 using YSAI.Core.script;
 using YSAI.Log;
 using YSAI.Unility;
+using static YSAI.Core.reflection.ReflectionData.Basics.DllData.NamespaceData.ClassData;
+using static YSAI.Core.reflection.ReflectionData.Basics.DllData.NamespaceData;
 
 namespace YSAI.Core.data
 {
@@ -49,12 +51,9 @@ namespace YSAI.Core.data
                 //原始值
                 string originalValue = value;
 
-                //解析s
+                //解析
                 value = Parse(addressDetails, value);
 
-                //异步转发
-                Relay(addressDetails, value);
-
                 //最终数据组织
                 AddressValue addressValue = new AddressValue().SET(addressDetails);
                 addressValue.AddressName = addressDetails.AddressName;
@@ -63,6 +62,9 @@ namespace YSAI.Core.data
                 addressValue.Time = DateTime.Now.ToLocalTime();
                 addressValue.Quality = 1;
 
+                //异步转发
+                Relay(addressValue);
+
                 return addressValue;
             }
             catch (Exception ex)
@@ -134,10 +136,10 @@ namespace YSAI.Core.data
         /// <summary>
         /// 数据转发
         /// </summary>
-        /// <param name="addressDetails">地址详情</param>
+        /// <param name="addressValue">地址详情</param>
         /// <param name="value">当前值</param>
         /// <returns></returns>
-        public static Task Relay(AddressDetails addressDetails, string value)
+        public static Task Relay(AddressValue addressValue)
         {
             return Task.Run(() =>
             {
@@ -146,39 +148,18 @@ namespace YSAI.Core.data
 
                 try
                 {
-                    //MQTT转发
-                    if (addressDetails.MqttRelay != null)
-                    {
-                        OperateResult? operateResult = reflectionOperate.ExecuteMethod(addressDetails.MqttRelay.ReflectionSN, new object[] { addressDetails.MqttRelay.Topic, value, addressDetails.MqttRelay.QoSLevel, addressDetails.MqttRelay.Retain }) as OperateResult;
-                        if (!operateResult.State)
-                        {
-                            DynamicObj.Param = addressDetails.MqttRelay;
-                            DynamicObj.State = operateResult.State;
-                            DynamicObj.Exception = operateResult.Message;
-                            LogHelper.Error(DynamicObj.ToJson(), logName);
-                        }
-                    }
-
-                    //RabbitMQ转发
-                    if (addressDetails.RabbitMQRelay != null)
-                    {
-                        OperateResult? operateResult = reflectionOperate.ExecuteMethod(addressDetails.RabbitMQRelay.ReflectionSN, new object[] { addressDetails.RabbitMQRelay.Topic, value, addressDetails.RabbitMQRelay.Type, addressDetails.RabbitMQRelay.Durable, addressDetails.RabbitMQRelay.Exclusive, addressDetails.RabbitMQRelay.AutoDelete }) as OperateResult;
-                        if (!operateResult.State)
-                        {
-                            DynamicObj.Param = addressDetails.RabbitMQRelay;
-                            DynamicObj.State = operateResult.State;
-                            DynamicObj.Exception = operateResult.Message;
-                            LogHelper.Error(DynamicObj.ToJson(), logName);
-                        }
-                    }
-
-                    //Kafka转发
-                    if (addressDetails.KafkaRelay != null)
+                    if (RegisterEvent())
                     {
-                        OperateResult? operateResult = reflectionOperate.ExecuteMethod(addressDetails.KafkaRelay.ReflectionSN, new object[] { addressDetails.KafkaRelay.Topic, addressDetails.KafkaRelay.Key, value }) as OperateResult;
+                        //类的唯一标识符
+                        string ClassSN = "YSAI.Relay.RelayOperate[Instance]";
+                        //方法的唯一标识符
+                        string MethodSN = "[Produce]";
+                        //执行转发方法 主题设置为空，走配置主题
+                        OperateResult operateResult = reflectionOperate.ExecuteMethod($"{ClassSN}{MethodSN}", new object[] { PacketType.Message.ToString(), addressValue.ToJson().JsonFormatting() }) as OperateResult;
+                        //状态判断
                         if (!operateResult.State)
                         {
-                            DynamicObj.Param = addressDetails.KafkaRelay;
+                            DynamicObj.Value = addressValue;
                             DynamicObj.State = operateResult.State;
                             DynamicObj.Exception = operateResult.Message;
                             LogHelper.Error(DynamicObj.ToJson(), logName);
@@ -192,5 +173,44 @@ namespace YSAI.Core.data
             });
         }
 
+        /// <summary>
+        /// 注册事件状态
+        /// </summary>
+        private static bool RegisterEventState = false;
+        /// <summary>
+        /// 执行注册事件
+        /// </summary>
+        /// <returns></returns>
+        private static bool RegisterEvent()
+        {
+            if (!RegisterEventState)
+            {
+                //类的唯一标识符
+                string ClassSN = "YSAI.Relay.RelayOperate[Instance]";
+                //事件的唯一标识符
+                string EventSN = "[OnEvent]";
+                //注册事件
+                OperateResult operateResult = reflectionOperate.RegisterEvent($"{ClassSN}{EventSN}", true, P2: RelayOperate_OnEvent);
+                if (!operateResult.State)
+                {
+                    throw new Exception($"注册事件失败：{operateResult.Message}");
+                }
+                RegisterEventState = operateResult.State;
+                return operateResult.State;
+            }
+            return true;
+        }
+
+        /// <summary>
+        /// 转发的事件注册
+        /// </summary>
+        /// <param name="sender"></param>
+        /// <param name="e"></param>
+        /// <exception cref="NotImplementedException"></exception>
+        private static void RelayOperate_OnEvent(object? sender, object e)
+        {
+            throw new NotImplementedException();
+        }
+
     }
 }

src/YSAI.DAQ/YSAI.Core/data/AddressRelay.cs

@@ -1,112 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using YSAI.Core.reflection;
-
-namespace YSAI.Core.data
-{
-    /// <summary>
-    /// 地址转发
-    /// </summary>
-    public class AddressRelay
-    {
-        /// <summary>
-        /// Mqtt转发数据
-        /// </summary>
-        public MqttRelayParam? MqttRelay { get; set; } = null;
-        /// <summary>
-        /// Kafka转发数据
-        /// </summary>
-        public KafkaRelayParam? KafkaRelay { get; set; } = null;
-        /// <summary>
-        /// RabbitMQ转发数据
-        /// </summary>
-        public RabbitMQRelayParam? RabbitMQRelay { get; set; } = null;
-        /// <summary>
-        /// MQTT转发数据
-        /// </summary>
-        public class MqttRelayParam
-        {
-            /// <summary>
-            /// 主题
-            /// </summary>
-            public string Topic { get; set; }
-            /// <summary>
-            /// 消息是否保留
-            /// </summary>
-            public bool Retain { get; set; } = false;
-            /// <summary>
-            /// 消息等级
-            /// QoS 0 是最低级别，基本上等同于 Fire and Forget 模式，发送者发送完数据之后，不关心消息是否已经投递到了接收者那边。
-            /// QoS 1 是中间级别，保证消息至少送达一次。MQTT 通过简单的 ACK 机制来保证 QoS1。
-            /// QoS 2 是最高级别，保证到且仅到一次。这通过更加复杂的消息流程保证。
-            /// 
-            /// 注意：QoS 级别越高，流程越复杂，系统资源消耗越大
-            /// </summary>
-            public int QoSLevel { get; set; } = 0;
-
-            /// <summary>
-            /// 通过反射得到的转发方法
-            /// </summary>
-            public string ReflectionSN { get; set; }
-        }
-        /// <summary>
-        /// Kafka生产者转发数据
-        /// </summary>
-        public class KafkaRelayParam
-        {
-            /// <summary>
-            /// 主题
-            /// </summary>
-            public string Topic { get; set; }
-            /// <summary>
-            /// 键
-            /// </summary>
-            public string Key { get; set; }
-
-            /// <summary>
-            /// 通过反射得到的转发方法
-            /// </summary>
-            public string ReflectionSN { get; set; }
-        }
-        /// <summary>
-        /// RabbitMQ转发数据
-        /// </summary>
-        public class RabbitMQRelayParam
-        {
-            /// <summary>
-            /// 消息头
-            /// </summary>
-            public string Topic { get; set; }
-            /// <summary>
-            /// 通道类型
-            /// direct  //用于AMQP直接交换的交换类型。
-            /// fanout  //用于AMQP扇出交换的交换类型。
-            /// headers //用于AMQP报头交换的交换类型。
-            /// topic   //用于AMQP主题交换的交换类型。
-            /// </summary>
-            public string Type { get; set; } = "topic";
-            /// <summary>
-            /// 持久化
-            /// </summary>
-            public bool Durable { get; set; } = true;
-            /// <summary>
-            /// 独有的
-            /// </summary>
-            public bool Exclusive { get; set; } = false;
-            /// <summary>
-            /// 自动删除
-            /// </summary>
-            public bool AutoDelete { get; set; } = false;
-
-            /// <summary>
-            /// 通过反射得到的转发方法
-            /// </summary>
-            public string ReflectionSN { get; set; }
-        }
-    }
-
-    
-}

src/YSAI.DAQ/YSAI.Core/data/AddressValue.cs

@@ -40,7 +40,6 @@ namespace YSAI.Core.data
             this.AddressDescribe = addressDetails.AddressDescribe;
             this.AddressParseParam = addressDetails.AddressParseParam;
             this.AddressExtendParam = addressDetails.AddressExtendParam;
-            this.MqttRelay = addressDetails.MqttRelay;
             this.AddressDataType = addressDetails.AddressDataType;
             this.AddressType = addressDetails.AddressType;
             return this;

src/YSAI.DAQ/YSAI.Core/data/packet/AddressPacket.cs

@@ -9,5 +9,5 @@ namespace YSAI.Core.data.packet
     /// <summary>
     /// 地址数据包（发送）
     /// </summary>
-    public class AddressPacket : AddressValue { }
+    public class MessagePacket : AddressValue { }
 }

src/YSAI.DAQ/YSAI.Core/data/packet/OperatePacket.cs

@@ -1,31 +0,0 @@
-using Newtonsoft.Json;
-using Newtonsoft.Json.Converters;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using YSAI.Core.@enum;
-
-namespace YSAI.Core.data.packet
-{
-    /// <summary>
-    /// 操作数据包（接收）
-    /// </summary>
-    public class OperatePacket
-    {
-        /// <summary>
-        /// 操作类型
-        /// </summary>
-        [JsonConverter(typeof(StringEnumConverter))]
-        public OperateType OType { get; set; }
-        /// <summary>
-        /// 操作传入的参数
-        /// </summary>
-        public object[]? Param { get; set; }
-        /// <summary>
-        /// 时间
-        /// </summary>
-        public DateTime Time { get; set; }
-    }
-}

src/YSAI.DAQ/YSAI.Core/enum/Enums.cs

@@ -191,6 +191,8 @@ namespace YSAI.Core.@enum
         [Description("S7采集")]
         S7
     }
+
+
     /// <summary>
     /// 转发类型
     /// </summary>
@@ -212,6 +214,8 @@ namespace YSAI.Core.@enum
         [Description("RabbitMQ转发")]
         RabbitMQ
     }
+
+
     /// <summary>
     /// 库类型
     /// </summary>
@@ -283,4 +287,20 @@ namespace YSAI.Core.@enum
         Off
     }
 
+    /// <summary>
+    /// 数据包类型
+    /// </summary>
+    public enum PacketType
+    {
+        /// <summary>
+        /// 消息
+        /// </summary>
+        [Description("消息")]
+        Message,
+        /// <summary>
+        /// 状态
+        /// </summary>
+        [Description("状态")]
+        Status
+    }
 }

src/YSAI.DAQ/YSAI.Core/interface/only/IKafka.cs

@@ -4,6 +4,7 @@ using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using YSAI.Core.data;
+using YSAI.Core.@interface.unify;
 
 namespace YSAI.Core.@interface.only
 {
@@ -60,7 +61,7 @@ namespace YSAI.Core.@interface.only
     /// <summary>
     /// 生产者接口
     /// </summary>
-    public interface IKafkaProducer
+    public interface IKafkaProducer:IProducer
     {
         /// <summary>
         /// 生产

src/YSAI.DAQ/YSAI.Core/interface/only/IMqtt.cs

@@ -12,7 +12,7 @@ namespace YSAI.Core.@interface.only
     /// <summary>
     /// MQTT客户端接口
     /// </summary>
-    public interface IMqttClient : IOn, IOff, IDisposable
+    public interface IMqttClient : IOn, IOff,IProducer, IDisposable
     {
         /// <summary>
         /// 发布订阅

src/YSAI.DAQ/YSAI.Core/interface/only/IRabbitMQ.cs

@@ -11,7 +11,7 @@ namespace YSAI.Core.@interface.only
     /// <summary>
     /// RabbitMQ接口
     /// </summary>
-    public interface IRabbitMQ : IOn, IOff, IDisposable
+    public interface IRabbitMQ : IOn, IOff, IProducer, IDisposable
     {
         /// <summary>
         /// 发布 一个交换机可以发布多个消息

src/YSAI.DAQ/YSAI.Core/reflection/Reflection.json

@@ -1,123 +0,0 @@
-{
-    "DllDatas": [
-        {
-            "DllPath": "YSAI.Mqtt.dll",
-            "IsAbsolutePath": false,
-            "NamespaceDatas": [
-                {
-                    "Namespace": "YSAI.Mqtt.client",
-                    "ClassDatas": [
-                        {
-                            "SN": "YSAI.Mqtt.MqttClientOperate[Instance]",
-                            "ClassName": "MqttClientOperate",
-                            "ConstructorParam": [
-                                {
-                                    "SN": "YSAI",
-                                    "ServerIPAddress": "127.0.0.1",
-                                    "ServerPort": 11819,
-                                    "ServerLoginID": "ysai",
-                                    "ServerLoginPassword": "ysai",
-                                    "ClientID": null,
-                                    "QualityOfServiceLevel": "AtMostOnce"
-                                }
-                            ],
-                            "MethodDatas": [
-                                {
-                                    "SN": "[Init]",
-                                    "WhetherExecute": true,
-                                    "MethodName": "Init",
-                                    "MethodParam": null
-                                },
-                                {
-                                    "SN": "[On]",
-                                    "WhetherExecute": true,
-                                    "MethodName": "On",
-                                    "MethodParam": null
-                                },
-                                {
-                                    "SN": "[PublishSubscribe]",
-                                    "WhetherExecute": false,
-                                    "MethodName": "PublishSubscribe",
-                                    "MethodParam": null
-                                }
-                            ],
-                            "EventDatas": null
-                        }
-                    ]
-                }
-            ]
-        },
-        {
-            "DllPath": "YSAI.Kafka.dll",
-            "IsAbsolutePath": false,
-            "NamespaceDatas": [
-                {
-                    "Namespace": "YSAI.Kafka",
-                    "ClassDatas": [
-                        {
-                            "SN": "YSAI.Kafka.KafkaProducerOperate[Instance]",
-                            "ClassName": "KafkaProducerOperate",
-                            "ConstructorParam": [
-                                {
-                                    "ClientId": "ysai",
-                                    "WaitTime": 1000,
-                                    "SN": "YSAI",
-                                    "BootstrapServers": "127.0.0.1:8083"
-                                }
-                            ],
-                            "MethodDatas": [
-                                {
-                                    "SN": "[Produce]",
-                                    "WhetherExecute": false,
-                                    "MethodName": "Produce",
-                                    "MethodParam": null
-                                }
-                            ],
-                            "EventDatas": null
-                        }
-                    ]
-                }
-            ]
-        },
-        {
-            "DllPath": "YSAI.RabbitMQ.dll",
-            "IsAbsolutePath": false,
-            "NamespaceDatas": [
-                {
-                    "Namespace": "YSAI.RabbitMQ",
-                    "ClassDatas": [
-                        {
-                            "SN": "YSAI.RabbitMQ.RabbitMQPublisherOperate[Instance]",
-                            "ClassName": "RabbitMQPublisherOperate",
-                            "ConstructorParam": [
-                                {
-                                    "SN": "YSAI",
-                                    "ExChangeName": "exYSAI",
-                                    "HostName": "localhost",
-                                    "Port": 4369,
-                                    "UserName": "ysai",
-                                    "Password": "ysai"
-                                }
-                            ],
-                            "MethodDatas": [
-                                {
-                                    "SN": "[On]",
-                                    "WhetherExecute": true,
-                                    "MethodName": "On",
-                                    "MethodParam": null
-                                },
-                                {
-                                    "SN": "[Publish]",
-                                    "WhetherExecute": false,
-                                    "MethodName": "Publish",
-                                    "MethodParam": null
-                                }
-                            ],
-                            "EventDatas": null
-                        }
-                    ]
-                }
-            ]
-        }
-    ]
-}

src/YSAI.DAQ/YSAI.Core/reflection/ReflectionOperate.cs

@@ -35,7 +35,7 @@ namespace YSAI.Core.reflection
         /// <summary>
         /// 文件
         /// </summary>
-        public static string ConfigFile = $"{AppDomain.CurrentDomain.BaseDirectory}{ConfigDirectory}//RelayConfig.json";
+        public static string ConfigFile = $"{AppDomain.CurrentDomain.BaseDirectory}{ConfigDirectory}//ReflectionConfig.json";
         /// <summary>
         /// 配置数据
         /// </summary>

src/YSAI.DAQ/YSAI.DAQ.sln

@@ -61,14 +61,14 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Manage.Windows", "YSAI
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "relay", "relay", "{9D8EDBBA-7A97-4D84-9B12-7FCC2F834046}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "YSAI.Relay", "YSAI.Relay\YSAI.Relay.csproj", "{E43BECDA-DE1F-446A-96E7-F110977597F2}"
-EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "core", "core", "{D60224CF-7F12-453B-851E-B5C01F9D2BBE}"
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "test", "test", "{12CB0510-7B1E-4518-AA3B-412A4D323D42}"
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "manage", "manage", "{ECAD410C-2895-4836-BCA7-D4EF340E778E}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Relay", "YSAI.Relay\YSAI.Relay.csproj", "{B8024157-E8D9-4CA8-941F-CAB610FAB046}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -179,10 +179,10 @@ Global
 		{2221CE78-FA24-40E3-8453-9C66E32F9C0C}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{2221CE78-FA24-40E3-8453-9C66E32F9C0C}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{2221CE78-FA24-40E3-8453-9C66E32F9C0C}.Release|Any CPU.Build.0 = Release|Any CPU
-		{E43BECDA-DE1F-446A-96E7-F110977597F2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{E43BECDA-DE1F-446A-96E7-F110977597F2}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{E43BECDA-DE1F-446A-96E7-F110977597F2}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{E43BECDA-DE1F-446A-96E7-F110977597F2}.Release|Any CPU.Build.0 = Release|Any CPU
+		{B8024157-E8D9-4CA8-941F-CAB610FAB046}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{B8024157-E8D9-4CA8-941F-CAB610FAB046}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{B8024157-E8D9-4CA8-941F-CAB610FAB046}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{B8024157-E8D9-4CA8-941F-CAB610FAB046}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -215,7 +215,7 @@ Global
 		{257F1474-B220-4C61-88C6-5B83BEF7B3A7} = {D60224CF-7F12-453B-851E-B5C01F9D2BBE}
 		{49133ADB-D3BF-4682-AA5A-CC1CF1917D45} = {D60224CF-7F12-453B-851E-B5C01F9D2BBE}
 		{2221CE78-FA24-40E3-8453-9C66E32F9C0C} = {ECAD410C-2895-4836-BCA7-D4EF340E778E}
-		{E43BECDA-DE1F-446A-96E7-F110977597F2} = {9D8EDBBA-7A97-4D84-9B12-7FCC2F834046}
+		{B8024157-E8D9-4CA8-941F-CAB610FAB046} = {9D8EDBBA-7A97-4D84-9B12-7FCC2F834046}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {5D5D3927-6714-40C0-84EA-44C5BA4C5E87}

src/YSAI.DAQ/YSAI.Kafka/KafkaProducerData.cs

@@ -24,6 +24,9 @@ namespace YSAI.Kafka
             /// </summary>
             [Description("等待时间")]
             public int WaitTime { get; set; } = 1000;
+
+           
+
             /// <summary>
             /// 重写基类中的Equals方法
             /// </summary>

src/YSAI.DAQ/YSAI.Kafka/KafkaProducerOperate.cs

@@ -133,5 +133,17 @@ namespace YSAI.Kafka
         {
             return Task.Run(() => Produce(Topic, Key, Content));
         }
+
+        public OperateResult Produce(string Topic, string Content)
+        {
+            Depart("Produce");
+            OperateResult operateResult = Produce(Topic, null, Content);
+            return Break("Produce", operateResult.State, operateResult.Message);
+        }
+
+        public Task<OperateResult> ProduceAsync(string Topic, string Content)
+        {
+            return Task.Run(() => Produce(Topic, Content));
+        }
     }
 }

src/YSAI.DAQ/YSAI.Mqtt/YSAI.Mqtt.csproj

@@ -13,6 +13,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\YSAI.Core\YSAI.Core.csproj" />
+    <ProjectReference Include="..\YSAI.Unility\YSAI.Unility.csproj" />
   </ItemGroup>
 
 </Project>

src/YSAI.DAQ/YSAI.Mqtt/client/MqttClientOperate.cs

@@ -6,8 +6,6 @@ using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using YSAI.Core.data;
-using YSAI.Unility;
-using YSAI.Log;
 using MQTTnet.Protocol;
 using static YSAI.Mqtt.client.MqttClientData;
 using System.Collections.Concurrent;
@@ -258,5 +256,17 @@ namespace YSAI.Mqtt.client
                 return Break("Off", false, ex.Message);
             }
         }
+
+        public OperateResult Produce(string Topic, string Content)
+        {
+            Depart("Produce");
+            OperateResult operateResult = PublishSubscribe(Topic, Content, Retain:true);
+            return Break("Produce", operateResult.State, operateResult.Message);
+        }
+
+        public Task<OperateResult> ProduceAsync(string Topic, string Content)
+        {
+            return Task.Run(() => Produce(Topic, Content));
+        }
     }
 }

src/YSAI.DAQ/YSAI.Mqtt/service/MqttServiceOperate.cs

@@ -10,8 +10,6 @@ using System.Net;
 using System.Text;
 using System.Threading.Tasks;
 using YSAI.Core.data;
-using YSAI.Log;
-using YSAI.Unility;
 using static YSAI.Mqtt.service.MqttServiceData;
 using YSAI.Core.@interface.only;
 using YSAI.Core.@interface.unify;

src/YSAI.DAQ/YSAI.Mqtt/service/websocket/MqttWebSocketServiceOperate.cs

@@ -6,12 +6,11 @@ using MQTTnet.Server;
 using MQTTnet.AspNetCore;
 using System.Text;
 using System;
-using YSAI.Unility;
 using YSAI.Core.data;
-using YSAI.Log;
 using static YSAI.Mqtt.service.websocket.MqttWebSocketServiceData;
 using YSAI.Core.@interface.only;
 using YSAI.Core.@interface.unify;
+using YSAI.Unility;
 
 namespace YSAI.Mqtt.service.websocket
 {

src/YSAI.DAQ/YSAI.RabbitMQ/RabbitMQData.cs

@@ -48,6 +48,7 @@ namespace YSAI.RabbitMQ
             [Description("密码")]
             public string? Password { get; set; } = null;
 
+
             /// <summary>
             /// 重写Equals
             /// </summary>
@@ -79,11 +80,11 @@ namespace YSAI.RabbitMQ
             /// <summary>
             /// 消息头
             /// </summary>
-            public string MessageHead { get; set; }
+            public string Topic { get; set; }
             /// <summary>
             /// 消息的内容
             /// </summary>
-            public string MessageContent { get; set; }
+            public string Content { get; set; }
 
             /// <summary>
             /// 时间

src/YSAI.DAQ/YSAI.RabbitMQ/RabbitMQOperate.cs

@@ -279,12 +279,24 @@ namespace YSAI.RabbitMQ
         /// </summary>
         private void Consumer_Received(object? sender, BasicDeliverEventArgs e, bool AutoAck = false)
         {
-            OnEventHandler?.Invoke(this, new RabbitMQData.Event() { MessageContent = Encoding.UTF8.GetString(e.Body.ToArray()), MessageHead = e.RoutingKey });
+            OnEventHandler?.Invoke(this, new RabbitMQData.Event() { Content = Encoding.UTF8.GetString(e.Body.ToArray()), Topic = e.RoutingKey });
             if (!AutoAck)
             {
                 //当自动确认为false，得手动确认消息
                 Channels[e.RoutingKey].BasicAck(e.DeliveryTag, false);
             }
         }
+
+        public OperateResult Produce(string Topic, string Content)
+        {
+            Depart("Produce");
+            OperateResult operateResult = Publish(Topic,Content);
+            return Break("Produce", operateResult.State, operateResult.Message);
+        }
+
+        public Task<OperateResult> ProduceAsync(string Topic, string Content)
+        {
+            return Task.Run(() => Produce(Topic, Content));
+        }
     }
 }

src/YSAI.DAQ/YSAI.Relay/RelayData.cs

@@ -15,6 +15,28 @@ namespace YSAI.Relay
         /// </summary>
         public class Basics
         {
+
+            /// <summary>
+            /// 消息主题
+            /// </summary>
+            [Description("消息主题")]
+            public string MessageTopic { get; set; } = "message";
+            /// <summary>
+            /// 状态主题
+            /// </summary>
+            [Description("状态主题")]
+            public string StatusTopic { get; set; } = "status";
+            /// <summary>
+            /// 任务处理数量
+            /// </summary>
+            [Description("任务处理数量")]
+            public int TaskHandleCount { get; set; } = 8;
+            /// <summary>
+            /// 任务处理完成休眠时间(毫秒)
+            /// </summary>
+            [Description("任务处理完成休眠时间(毫秒)")]
+            public int TaskHandleAccomplishSleepTime { get; set; } = 500;
+
             /// <summary>
             /// MQTT参数集合
             /// </summary>
@@ -24,13 +46,10 @@ namespace YSAI.Relay
             /// </summary>
             public List<KafkaProducerData.Basics>? KafkaProducerDataArray { get; set; }
             /// <summary>
-            /// Kafka消费者参数集合
-            /// </summary>
-            public List<KafkaConsumerData.Basics>? KafkaConsumerDataArray { get; set; }
-            /// <summary>
             /// RabbitMQ 参数集合
             /// </summary>
             public List<RabbitMQData.Basics>? RabbitMQDataArray { get; set; }
+
             /// <summary>
             /// 重写基类中的Equals方法
             /// </summary>
@@ -62,5 +81,6 @@ namespace YSAI.Relay
                 }
             }
         }
+
     }
 }

src/YSAI.DAQ/YSAI.Relay/RelayOperate.cs

@@ -5,111 +5,327 @@ using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using YSAI.Core.data;
+using YSAI.Core.@enum;
 using YSAI.Core.@interface.unify;
+using YSAI.Core.reflection;
 using YSAI.Kafka;
+using YSAI.Log;
 using YSAI.Mqtt.client;
 using YSAI.RabbitMQ;
+using YSAI.Unility;
 
 namespace YSAI.Relay
 {
     /// <summary>
     /// 转发操作
     /// </summary>
-    public class RelayOperate:IBaseAbstract,IRelay
+    public class RelayOperate : IBaseAbstract<object>, IRelay
     {
         protected override string LogHead => "[ RelayOperate 操作 ]";
         protected override string ClassName => "RelayOperate";
+        /// <summary>
+        /// 文件夹
+        /// </summary>
+        public static string ConfigDirectory = "config";
+        /// <summary>
+        /// 文件
+        /// </summary>
+        public static string ConfigFile = $"{AppDomain.CurrentDomain.BaseDirectory}{ConfigDirectory}//RelayConfig.json";
+
+        /// <summary>
+        /// 统一返回
+        /// </summary>
+        private OperateResult? operateResult = null;
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        private RelayData.Basics basics;
 
         private static readonly object Lock = new object();  //锁
-        private static List<RelayOperate> ThisObjList = new List<RelayOperate>(); //自身对象集合
+        private static ReflectionOperate? ThisObjList; //自身对象
         /// <summary>
         /// 单例模式
         /// </summary>
         /// <returns></returns>
-        public static RelayOperate Instance(RelayData.Basics basics)
+        public static ReflectionOperate Instance()
         {
-            RelayOperate? exp = ThisObjList.FirstOrDefault(c => c.basics.Equals(basics));
-            if (exp == null)
+            if (ThisObjList == null)
             {
                 lock (Lock)
                 {
-                    if (ThisObjList.Count(c => c.basics.Equals(basics)) > 0)
+                    if (ThisObjList == null)
                     {
-                        return ThisObjList.First(c => c.basics.Equals(basics));
-                    }
-                    else
-                    {
-                        RelayOperate exp2 = new RelayOperate(basics);
-                        ThisObjList.Add(exp2);
-                        return exp2;
+                        ThisObjList = new ReflectionOperate();
                     }
                 }
             }
-            return exp;
+            return ThisObjList;
         }
-        /// <summary>
-        /// kafka消费者
-        /// </summary>
-        public ConcurrentDictionary<string, KafkaConsumerOperate> KafkaConsumerArray;
+
         /// <summary>
         /// kafka生产者
         /// </summary>
-        public ConcurrentDictionary<string, KafkaProducerOperate> KafkaProducerArray;
+        private ConcurrentDictionary<string, KafkaProducerOperate> KafkaProducerArray;
         /// <summary>
         /// MQTT
         /// </summary>
-        public ConcurrentDictionary<string, MqttClientOperate> MqttClientArray;
+        private ConcurrentDictionary<string, MqttClientOperate> MqttClientArray;
         /// <summary>
         /// RabbitMQ
         /// </summary>
-        public ConcurrentDictionary<string, RabbitMQOperate> RabbitMQArray;
-
-
+        private ConcurrentDictionary<string, RabbitMQOperate> RabbitMQArray;
 
+        /// <summary>
+        /// 任务集合
+        /// </summary>
+        public ConcurrentDictionary<string, Task> TaskArray;
+        /// <summary>
+        /// 数据队列
+        /// </summary>
+        private ConcurrentQueue<QueueData> DataQueue;
+        /// <summary>
+        /// 队列里面的数据
+        /// </summary>
+        public class QueueData
+        {
+            /// <summary>
+            /// 主题
+            /// </summary>
+            public string Topic { get; set; }
+            /// <summary>
+            /// 内容
+            /// </summary>
+            public string Content { get; set; }
+        }
 
+        /// <summary>
+        /// 任务处理
+        /// </summary>
+        /// <returns></returns>
+        public Task TaskHandle()
+        {
+            string logName = "TaskHandle.log";
+            //起个新线程处理
+            return Task.Factory.StartNew(() =>
+            {
+                //循环
+                while (true)
+                {
+                    try
+                    {
+                        //队列数据
+                        QueueData? queueData;
+                        //出列
+                        while (DataQueue.TryDequeue(out queueData))
+                        {
+                            if (queueData != null)
+                            {
+                                if (KafkaProducerArray != null)
+                                {
+                                    foreach (var operate in KafkaProducerArray)
+                                    {
+                                        OnEventHandler?.Invoke(this, operate.Value.Produce(queueData.Topic, queueData.Content));
+                                    }
+                                }
+                                if (MqttClientArray != null)
+                                {
+                                    foreach (var operate in MqttClientArray)
+                                    {
+                                        OnEventHandler?.Invoke(this, operate.Value.Produce(queueData.Topic, queueData.Content));
+                                    }
+                                }
+                                if (RabbitMQArray != null)
+                                {
+                                    foreach (var operate in RabbitMQArray)
+                                    {
+                                        OnEventHandler?.Invoke(this, operate.Value.Produce(queueData.Topic, queueData.Content));
+                                    }
+                                }
+                            }
+                        }
+                    }
+                    catch (Exception ex)
+                    {
+                        LogHelper.Error($"任务处理异常：{ex.Message}", logName);
+                    }
+                    Thread.Sleep(basics.TaskHandleAccomplishSleepTime);
+                }
+            });
+        }
 
         /// <summary>
-        /// 构造函数
+        /// 无惨构造函数
         /// </summary>
-        /// <param name="basics"></param>
-        public RelayOperate(RelayData.Basics basics)
+        public RelayOperate()
         {
-            this.basics = basics;
-            //当进入构造函数直接初始化
+            this.basics = GetConfig();
+            Init();
+        }
 
+        /// <summary>
+        /// 获取配置
+        /// </summary>
+        /// <returns></returns>
+        private RelayData.Basics? GetConfig()
+        {
+            try
+            {
+                if (File.Exists(ConfigFile))  //配置存在
+                {
+                    return JsonTool.StringToJsonEntity<RelayData.Basics>(FileTool.FileToString(ConfigFile));
+                }
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine($"获取配置异常:{ex.Message}");
+            }
+            return null;
         }
 
         /// <summary>
-        /// 基础数据
+        /// 初始化状态
         /// </summary>
-        private RelayData.Basics basics;
+        private bool InitState = false;
+
+        /// <summary>
+        /// 初始化
+        /// </summary>
+        private void Init()
+        {
+            if (basics != null)
+            {
+
+                if (basics.RabbitMQDataArray != null)
+                {
+                    foreach (var configData in basics.RabbitMQDataArray)
+                    {
+                        RabbitMQOperate operate = RabbitMQOperate.Instance(configData);
+                        operateResult = operate.On();
+                        if (operateResult.State)
+                        {
+                            RabbitMQArray.TryAdd(configData.SN, operate);
+                        }
+                        else
+                        {
+                            //通知到外部，让外部来觉得是否重新初始化
+                            OnEventHandler?.Invoke(this, operateResult);
+                        }
+                    }
+                }
+
+                if (basics.MqttClientDataArray != null)
+                {
+                    foreach (var configData in basics.MqttClientDataArray)
+                    {
+                        MqttClientOperate operate = MqttClientOperate.Instance(configData);
+                        operateResult = operate.On();
+                        if (operateResult.State)
+                        {
+                            MqttClientArray.TryAdd(configData.SN, operate);
+                        }
+                        else
+                        {
+                            //通知到外部，让外部来觉得是否重新初始化
+                            OnEventHandler?.Invoke(this, operateResult);
+                        }
+                    }
+                }
+
+                if (basics.KafkaProducerDataArray != null)
+                {
+                    foreach (var configData in basics.KafkaProducerDataArray)
+                    {
+                        KafkaProducerOperate operate = KafkaProducerOperate.Instance(configData);
+                        KafkaProducerArray.TryAdd(configData.SN, operate);
+                    }
+                }
+
+                InitState = true;
 
+            }
+            else
+            {
+                OnEventHandler?.Invoke(this, Break(Depart("Init"), false, "配置文件不存在"));
+            }
+        }
 
         public OperateResult Produce(string Topic, string Content)
         {
-            throw new NotImplementedException();
+            Depart("Produce");
+            try
+            {
+                //主题
+                string topic = Topic;
+                try
+                {
+                    //如果包类型可以被转换，直接走对应配置，如果不能，则走传入的主题
+                    PacketType? packetType = (PacketType)Enum.Parse(typeof(PacketType), Topic);
+                    if (packetType != null)
+                    {
+                        switch (packetType)
+                        {
+                            case PacketType.Message:
+                                topic = basics.MessageTopic;
+                                break;
+                            case PacketType.Status:
+                                topic = basics.StatusTopic;
+                                break;
+                        }
+                    }
+                }
+                catch { }
+                if (!InitState)
+                {
+                    return Break("Produce", false, "尚未初始化");
+                }
+                else
+                {
+                    //当队列为空，初始化队列
+                    if (DataQueue == null)
+                    {
+                        DataQueue = new ConcurrentQueue<QueueData>();
+                        TaskArray = new ConcurrentDictionary<string, Task>();
+                        //启动
+                        for (int i = 0; i < basics.TaskHandleCount; i++)
+                        {
+                            TaskArray.TryAdd(Guid.NewGuid().ToString(), TaskHandle());
+                        }
+                    }
+                    //入列
+                    DataQueue.Enqueue(new QueueData() { Topic = Topic, Content = Content });
+                }
+                return Break("Produce", true);
+            }
+            catch (Exception ex)
+            {
+                return Break("Produce", false, ex.Message);
+            }
         }
 
         public Task<OperateResult> ProduceAsync(string Topic, string Content)
         {
-            throw new NotImplementedException();
+            return Task.Run(() => Produce(Topic, Content));
         }
 
         public OperateResult Consume(string Topic)
         {
-            throw new NotImplementedException();
+            return Break(Depart("Consume"), false, "目前暂不支持消费");
         }
 
         public Task<OperateResult> ConsumeAsync(string Topic)
         {
-            throw new NotImplementedException();
+            return Task.Run(() => Consume(Topic));
         }
 
         public void Dispose()
         {
-            throw new NotImplementedException();
+            if (KafkaProducerArray != null) { foreach (var item in KafkaProducerArray) { item.Value.Dispose(); } KafkaProducerArray = null; }
+            if (MqttClientArray != null) { foreach (var item in MqttClientArray) { item.Value.Dispose(); } MqttClientArray = null; }
+            if (RabbitMQArray != null) { foreach (var item in RabbitMQArray) { item.Value.Dispose(); } RabbitMQArray = null; }
+            if (TaskArray != null) { foreach (var item in TaskArray) { item.Value.Dispose(); } TaskArray = null; }
+            ThisObjList = null;
+            GC.Collect();
+            GC.SuppressFinalize(this);
         }
-
-
     }
 }

src/YSAI.DAQ/YSAI.Relay/YSAI.Relay.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFramework>net6.0</TargetFramework>
@@ -7,5 +7,17 @@
   </PropertyGroup>
 	<ItemGroup>
 		<ProjectReference Include="..\YSAI.Core\YSAI.Core.csproj" />
+		<ProjectReference Include="..\YSAI.Kafka\YSAI.Kafka.csproj" />
+		<ProjectReference Include="..\YSAI.Mqtt\YSAI.Mqtt.csproj" />
+		<ProjectReference Include="..\YSAI.RabbitMQ\YSAI.RabbitMQ.csproj" />
+		<ProjectReference Include="..\YSAI.Unility\YSAI.Unility.csproj" />
+	</ItemGroup>
+	<ItemGroup>
+	  <None Update="config\ReflectionConfig.json">
+	    <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+	  </None>
+	  <None Update="config\RelayConfig.json">
+	    <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+	  </None>
 	</ItemGroup>
 </Project>

src/YSAI.DAQ/YSAI.Relay/config/ReflectionConfig.json

@@ -0,0 +1,40 @@
+{
+    "DllDatas": [
+        {
+            "DllPath": "YSAI.Relay.dll",
+            "IsAbsolutePath": false,
+            "NamespaceDatas": [
+                {
+                    "Namespace": "YSAI.Relay",
+                    "ClassDatas": [
+                        {
+                            "SN": "YSAI.Relay.RelayOperate[Instance]",
+                            "ClassName": "RelayOperate",
+                            "ConstructorParam": null,
+                            "MethodDatas": [
+                                {
+                                    "SN": "[Produce]",
+                                    "WhetherExecute": false,
+                                    "MethodName": "Produce",
+                                    "MethodParam": null
+                                },
+                                {
+                                    "SN": "[Dispose]",
+                                    "WhetherExecute": false,
+                                    "MethodName": "Dispose",
+                                    "MethodParam": null
+                                }
+                            ],
+                            "EventDatas": [
+                                {
+                                    "SN": "[OnEvent]",
+                                    "EventName": "OnEvent"
+                                }
+                            ]
+                        }
+                    ]
+                }
+            ]
+        }
+    ]
+}

src/YSAI.DAQ/YSAI.Relay/config/RelayConfig.json

@@ -0,0 +1,44 @@
+{
+    "MessageTopic": "message",
+    "StatusTopic": "status",
+    "MqttClientDataArray": [
+        {
+            "SN": "mqtt转发一号",
+            "ServerIPAddress": "127.0.0.1",
+            "ServerPort": 8881,
+            "ServerLoginID": "ysai",
+            "ServerLoginPassword": "ysai",
+            "ClientID": null,
+            "QualityOfServiceLevel": "AtMostOnce"
+        },
+        {
+            "SN": "mqtt转发二号",
+            "ServerIPAddress": "127.0.0.1",
+            "ServerPort": 8882,
+            "ServerLoginID": "ysai",
+            "ServerLoginPassword": "ysai",
+            "ClientID": null,
+            "QualityOfServiceLevel": "AtMostOnce"
+        },
+        {
+            "SN": "mqtt转发三号",
+            "ServerIPAddress": "127.0.0.1",
+            "ServerPort": 8883,
+            "ServerLoginID": "ysai",
+            "ServerLoginPassword": "ysai",
+            "ClientID": null,
+            "QualityOfServiceLevel": "AtMostOnce"
+        },
+        {
+            "SN": "mqtt转发四号",
+            "ServerIPAddress": "127.0.0.1",
+            "ServerPort": 8884,
+            "ServerLoginID": "ysai",
+            "ServerLoginPassword": "ysai",
+            "ClientID": null,
+            "QualityOfServiceLevel": "AtMostOnce"
+        }
+    ],
+    "KafkaProducerDataArray": [],
+    "RabbitMQDataArray": []
+}

src/YSAI.DAQ/YSAI.TestConsole/Program.cs

@@ -23,6 +23,7 @@ using static YSAI.Core.reflection.ReflectionData;
 using static YSAI.Core.reflection.ReflectionData.Basics.DllData.NamespaceData;
 using YSAI.Core.@enum;
 using System.Text;
+using YSAI.Relay;
 
 namespace YSAI.TestConsole
 {
@@ -31,16 +32,87 @@ namespace YSAI.TestConsole
         static void Main(string[] args)
         {
 
+            RelayData.Basics basics = new RelayData.Basics();
+            basics.RabbitMQDataArray = new List<RabbitMQData.Basics>();
+            basics.KafkaProducerDataArray = new List<KafkaProducerData.Basics>();
+            basics.MqttClientDataArray = new List<MqttClientData.Basics>();
+            basics.MqttClientDataArray.Add(new MqttClientData.Basics() { QualityOfServiceLevel = 0, ServerIPAddress = "127.0.0.1", ServerPort = 8881, ServerLoginID = "ysai", ServerLoginPassword = "ysai", SN = "mqtt转发一号" });
+            basics.MqttClientDataArray.Add(new MqttClientData.Basics() { QualityOfServiceLevel = 0, ServerIPAddress = "127.0.0.1", ServerPort = 8882, ServerLoginID = "ysai", ServerLoginPassword = "ysai", SN = "mqtt转发二号" });
+            basics.MqttClientDataArray.Add(new MqttClientData.Basics() { QualityOfServiceLevel = 0, ServerIPAddress = "127.0.0.1", ServerPort = 8883, ServerLoginID = "ysai", ServerLoginPassword = "ysai", SN = "mqtt转发三号" });
+            basics.MqttClientDataArray.Add(new MqttClientData.Basics() { QualityOfServiceLevel = 0, ServerIPAddress = "127.0.0.1", ServerPort = 8884, ServerLoginID = "ysai", ServerLoginPassword = "ysai", SN = "mqtt转发四号" });
 
-           
+            RelayOperate relayOperate = new RelayOperate();
 
+            relayOperate.OnEvent += RelayOperate_OnEvent;
+
+            FileTool.StringToFile(RelayOperate.ConfigFile, basics.ToJson().JsonFormatting());
+
+
+
+            //创建配置
+            Basics basics2 = new Basics()
+            {
+                DllDatas = new List<Basics.DllData>()
+                {
+                    new Basics.DllData()
+                    {
+                        DllPath="YSAI.Relay.dll",
+                        IsAbsolutePath=false,
+                        NamespaceDatas=new List<Basics.DllData.NamespaceData>()
+                        {
+                            new Basics.DllData.NamespaceData()
+                            {
+                                Namespace="YSAI.Relay",
+                                ClassDatas=new List<Basics.DllData.NamespaceData.ClassData>()
+                                {
+                                    new Basics.DllData.NamespaceData.ClassData()
+                                    {
+                                        ClassName="RelayOperate",
+                                        ConstructorParam=null,
+                                        SN="YSAI.Relay.RelayOperate[Instance]",
+                                        MethodDatas=new List<Basics.DllData.NamespaceData.ClassData.MethodData>()
+                                        {
+                                            new Basics.DllData.NamespaceData.ClassData.MethodData()
+                                            {
+                                                MethodName="Produce",
+                                                SN="[Produce]"
+                                            },
+                                             new Basics.DllData.NamespaceData.ClassData.MethodData()
+                                            {
+                                                MethodName="Dispose",
+                                                SN="[Dispose]"
+                                            }
+                                        },
+                                        EventDatas=new List<ClassData.EventData>()
+                                        {
+                                            new ClassData.EventData()
+                                            {
+                                                EventName="OnEvent",
+                                                SN="[OnEvent]"
+                                            }
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            };
+
+
+            FileTool.StringToFile(ReflectionOperate.ConfigFile, basics2.ToJson().JsonFormatting());
+
+            Console.WriteLine();
 
             while (true)
             {
                 Console.ReadKey();
             }
         }
-       
 
+        private static void RelayOperate_OnEvent(object? sender, object e)
+        {
+            throw new NotImplementedException();
+        }
     }
 }

src/YSAI.DAQ/YSAI.TestConsole/YSAI.TestConsole.csproj

@@ -16,6 +16,7 @@
     <ProjectReference Include="..\YSAI.Opc\YSAI.Opc.csproj" />
     <ProjectReference Include="..\YSAI.RabbitMQ\YSAI.RabbitMQ.csproj" />
     <ProjectReference Include="..\YSAI.Redis\YSAI.Redis.csproj" />
+    <ProjectReference Include="..\YSAI.Relay\YSAI.Relay.csproj" />
     <ProjectReference Include="..\YSAI.S7\YSAI.S7.csproj" />
   </ItemGroup>