1

src/YSAI.DAQ/YSAI.Core/http/HttpData.cs

@@ -0,0 +1,136 @@
+using System.Collections.Specialized;
+using System.Net;
+using System.Text;
+
+namespace YSAI.Core.http
+{
+    /// <summary>
+    /// HTTP数据
+    /// </summary>
+    public class HttpData
+    {
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        public class Basics
+        {
+            /// <summary>
+            /// 唯一标识符
+            /// </summary>
+            public string SN { get; set; } = Guid.NewGuid().ToString();
+            /// <summary>
+            /// 最大连接数
+            /// </summary>
+            public int MaxConnectCount { get; set; } = 10000;
+
+            /// <summary>
+            /// 重写Equals
+            /// </summary>
+            /// <param name="obj"></param>
+            /// <returns></returns>
+            public override bool Equals(object Obj)
+            {
+                Basics? obj = Obj as Basics;
+                if (obj == null) return false;
+                if (obj.SN == this.SN&&obj.MaxConnectCount==this.MaxConnectCount)
+                {
+                    return true;
+                }
+                else
+                {
+                    return false;
+                }
+            }
+        }
+
+        /// <summary>
+        /// 请求的数据
+        /// </summary>
+        public class RequestData
+        {
+            public enum CType
+            {
+                FormData,
+                Raw
+            }
+            /// <summary>
+            /// 请求的链接
+            /// </summary>
+            public string Url { get; set; }
+            /// <summary>
+            /// 请求标头
+            /// </summary>
+            public Dictionary<string, string> Headers { get; set; } = new Dictionary<string, string>();
+            /// <summary>
+            /// 请求方式
+            /// </summary>
+            public HttpMethod Method { get; set; } = HttpMethod.Get;
+            /// <summary>
+            /// 内容数据类型
+            /// </summary>
+            public CType cType { get; set; } = CType.Raw;
+
+            /// <summary>
+            /// Form表单提交数据
+            /// </summary>
+            public Dictionary<string, string> FormDatas { get; set; } = new Dictionary<string, string>();
+
+            /// <summary>
+            /// 内容
+            /// </summary>
+            public string? Content { get; set; } = null;
+
+            /// <summary>
+            /// 编码格式
+            /// </summary>
+            public Encoding Encoding { get; set; } = Encoding.UTF8;
+            /// <summary>
+            /// 内容类型
+            /// </summary>
+            public string ContentType { get; set; } = "application/json";
+            /// <summary>
+            /// 超时时间
+            /// </summary>
+            public TimeSpan TimeOut { get; set; } = new TimeSpan(0, 0, 60);
+            /// <summary>
+            /// 代理
+            /// </summary>
+            public IWebProxy? Proxy { get; set; } = null;
+            /// <summary>
+            /// 缓存
+            /// </summary>
+            public CookieContainer CookieContainer { get; set; } = new CookieContainer();
+        }
+
+        /// <summary>
+        /// 响应数据
+        /// </summary>
+        public class ResponseData
+        {
+            /// <summary>
+            /// 状态码
+            /// </summary>
+            public int StatusCode { get; set; }
+
+            /// <summary>
+            /// 响应头数据
+            /// </summary>
+            public Dictionary<string, string>? ResHeadersData { get; set; }
+
+            /// <summary>
+            /// 响应的COOKIE
+            /// </summary>
+            public CookieCollection ResCookieData { get; set; } = new CookieCollection();
+
+            /// <summary>
+            /// 响应的数据
+            /// </summary>
+            public string? ResData { get; set; }
+
+            /// <summary>
+            /// 请求的数据
+            /// </summary>
+            public RequestData ReqData { get; set; }
+        }
+    }
+}

src/YSAI.DAQ/YSAI.Core/http/HttpOperate.cs

@@ -0,0 +1,188 @@
+using System.Net;
+using YSAI.Core.data;
+using YSAI.Core.@interface;
+using static YSAI.Core.http.HttpData;
+
+namespace YSAI.Core.http
+{
+    /// <summary>
+    /// HTTP操作
+    /// </summary>
+    public class HttpOperate : IBaseAbstract, IDisposable
+    {
+        protected override string LogHead => "[ HttpOperate 操作 ]";
+        protected override string ClassName => "HttpOperate";
+        private static readonly object Lock = new object();  //锁
+        private static List<HttpOperate> ThisObjList = new List<HttpOperate>(); //自身对象集合
+
+        /// <summary>
+        /// 单例模式
+        /// </summary>
+        /// <returns></returns>
+        public static HttpOperate Instance(HttpData.Basics basics)
+        {
+            HttpOperate? exp = ThisObjList.FirstOrDefault(c => c.basics.Equals(basics));
+            if (exp == null)
+            {
+                lock (Lock)
+                {
+                    if (ThisObjList.Count(c => c.basics.Equals(basics)) > 0)
+                    {
+                        return ThisObjList.First(c => c.basics.Equals(basics));
+                    }
+                    else
+                    {
+                        HttpOperate exp2 = new HttpOperate(basics);
+                        ThisObjList.Add(exp2);
+                        return exp2;
+                    }
+                }
+            }
+            return exp;
+        }
+
+        /// <summary>
+        /// 构造函数
+        /// </summary>
+        /// <param name="basics"></param>
+        public HttpOperate(HttpData.Basics basics)
+        {
+            this.basics = basics;
+        }
+
+        public HttpOperate()
+        {
+            this.basics = new HttpData.Basics();
+        }
+
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        private HttpData.Basics basics { get; set; }
+
+        /// <summary>
+        /// 请求
+        /// </summary>
+        /// <param name="Body">参数</param>
+        /// <param name="encoding">编码格式</param>
+        /// <returns></returns>
+        public OperateResult Request(RequestData requestData)
+        {
+            Depart("Request");
+            try
+            {
+                bool UseProxy = requestData.Proxy == null ? false : true;
+                bool UseCookies = requestData.CookieContainer.Count > 0 ? true : false;
+                using (HttpClientHandler httpClientHandler = new HttpClientHandler { UseCookies = UseCookies, CookieContainer = requestData.CookieContainer, UseProxy = UseProxy, Proxy = requestData.Proxy, MaxConnectionsPerServer = basics.MaxConnectCount })
+                using (HttpClient httpClient = new HttpClient(httpClientHandler))
+                {
+                    //设置超时时间
+                    httpClient.Timeout = requestData.TimeOut;
+
+                    //实例化对象，设置请求数据
+                    HttpRequestMessage request = new HttpRequestMessage(requestData.Method, requestData.Url);
+
+                    //设置请求头
+                    if (requestData.Headers != null)
+                    {
+                        foreach (var item in requestData.Headers)
+                        {
+                            request.Headers.Add(item.Key, item.Value);
+                        }
+                    }
+
+                    //内容赋值
+                    switch (requestData.cType)
+                    {
+                        case RequestData.CType.FormData:
+                            MultipartFormDataContent multipartFormDataContent = new MultipartFormDataContent();
+                            if (requestData.FormDatas != null)
+                            {
+                                foreach (var item in requestData.FormDatas)
+                                {
+                                    multipartFormDataContent.Add(new StringContent(item.Value), item.Key);
+                                }
+                            }
+                            if (multipartFormDataContent != null && multipartFormDataContent.Count() > 0)
+                            {
+                                multipartFormDataContent.Headers.ContentType.MediaType = requestData.ContentType;
+                                request.Content = multipartFormDataContent;
+                            }
+                            break;
+                        case RequestData.CType.Raw:
+                            //设置请求参数
+                            if (!string.IsNullOrEmpty(requestData.Content))
+                            {
+                                request.Content = new StringContent(requestData.Content, requestData.Encoding, requestData.ContentType);  //设置post请求参数
+                            }
+                            break;
+                    }
+                    //数据发送
+                    HttpResponseMessage response = httpClient.Send(request);
+                    //获取响应流
+                    StreamReader streamReader = new StreamReader(response.Content.ReadAsStream(), requestData.Encoding);
+                    //获取Cookie
+                    CookieCollection cookies = new CookieCollection();
+                    //转换响应头数据
+                    Dictionary<string, string>? ResponseHeadersData = DisposeResponseData(response.Headers.ToString(), ref cookies, requestData.Url);
+                    //响应数据
+                    string? ResponseData = streamReader.ReadToEnd();
+                    //状态码
+                    int StateCode = (int)response.StatusCode;
+                    //关闭
+                    streamReader.Close();
+                    //释放
+                    streamReader.Dispose();
+                    //返回数据
+                    return Break("Request", true, "请求成功", new HttpData.ResponseData() { ResData = ResponseData, ResHeadersData = ResponseHeadersData, StatusCode = StateCode, ReqData = requestData, ResCookieData = cookies }, Core.@enum.ResultType.Dynamic);
+                }
+            }
+            catch (Exception ex)
+            {
+                return Break("Request", false, ex.Message);
+            }
+        }
+
+        /// <summary>
+        /// 解析响应数据
+        /// </summary>
+        /// <param name="Data"></param>
+        /// <returns></returns>
+        private Dictionary<string, string>? DisposeResponseData(string Data,ref CookieCollection cookieContainer,string url)
+        {
+            string[] Datas = Data.Split(new char[] { '\r', '\n' });
+            if (Datas.Length > 0)
+            {
+                Dictionary<string,string> tuples = new Dictionary<string, string>();
+                foreach (var item in Datas)
+                {
+                    if (!string.IsNullOrEmpty(item))
+                    {
+                        string[] infos = item.Split(new char[] { ':' });
+                        tuples.Add(infos[0], infos[1]);
+                        if (infos[0].Equals("Set-Cookie"))
+                        {
+                            string[] strings = infos[1].Split(';');
+                            string[] KV1 = strings[0].Split('=');
+                            url = url.Replace("https:", "").Replace("http:", "").Replace("/", "");
+                            cookieContainer.Add(new Cookie(KV1[0].Trim(), KV1[1].Trim(), "/", url.Trim()));
+                        }
+
+                    }
+                }
+                return tuples;
+            }
+            return null;
+        }
+
+        /// <summary>
+        /// 释放
+        /// </summary>
+        public void Dispose()
+        {
+            GC.Collect();
+            GC.SuppressFinalize(this);
+            ThisObjList.Remove(this);
+        }
+    }
+}

src/YSAI.DAQ/YSAI.Kafka/KafkaData.cs

@@ -55,7 +55,7 @@ namespace YSAI.Kafka
             /// </summary>
             [Description("自动偏移复位")]
             [JsonConverter(typeof(StringEnumConverter))]
-            public AutoOffsetReset AutoOffsetReset { get; set; } = AutoOffsetReset.Earliest;
+            public AutoOffsetReset AutoOffsetReset { get; set; } = AutoOffsetReset.Latest;   
 
             /// <summary>
             /// 重写基类中的Equals方法

src/YSAI.DAQ/YSAI.Opc/ua/client/OpcUaClientOperate.cs

@@ -1068,7 +1068,7 @@ namespace YSAI.Opc.ua.client
                     UserIdentity = new UserIdentity(basics.UserName, basics.Password);
                 }
                 //通过调用CreateSession服务创建与服务器的新通信会话
-                clientSession = Session.Create(AC, endpoint, false, false, AC.ApplicationName, uint.MaxValue, UserIdentity, null).Result;
+                clientSession = Session.Create(AC, endpoint, false, false, AC.ApplicationName, 1800000, UserIdentity, null).Result;
 
                 //当会话关闭，则关闭订阅
                 clientSession.DeleteSubscriptionsOnClose = true;

src/YSAI.DAQ/YSAI.Test.Console/Program.cs

@@ -1,71 +1,97 @@
-//using Confluent.Kafka;
+using Confluent.Kafka;
 using S7.Net;
 using System.Collections.Concurrent;
 using YSAI.Core.data;
 using YSAI.Core.@enum;
 using YSAI.Core.reflection;
 using YSAI.Core.script;
-//using YSAI.DaqManage;
-//using YSAI.Log;
-//using YSAI.Mqtt.client;
-//using YSAI.Mqtt.service.websocket;
-//using YSAI.Opc.ua.client;
-//using YSAI.RabbitMQ;
-//using YSAI.RelayManage;
+using YSAI.DaqManage;
+using YSAI.Kafka;
+using YSAI.Log;
+using YSAI.Mqtt.client;
+using YSAI.Mqtt.service.websocket;
+using YSAI.Opc.ua.client;
+using YSAI.RabbitMQ;
+using YSAI.RelayManage;
 using YSAI.S7.client;
 using YSAI.Unility;
+using ZstdSharp.Unsafe;
 
-
-S7ClientOperate s7ClientOperate = S7ClientOperate.Instance(new S7ClientData.Basics()
+KafkaOperate kafkaOperate = KafkaOperate.Instance(new KafkaData.Basics() 
 {
-    Ip = "192.168.2.20",
-    Port = 102,
-    Rack = 0,
-    Slot = 1,
-    S7CpuType = CpuType.S71200,
-    SN = "1",
-    DataChangeOut=true
+    BootstrapServers= "192.168.2.140:9092",
 });
 
-//点位地址
-Address address = new Address();
-address.SN = Guid.NewGuid().ToString();
-address.CreationTime = DateTime.Now.ToLocalTime();
-address.AddressArray = new List<AddressDetails>();
-address.AddressArray.Add(new AddressDetails()
+OperateResult operateResult= kafkaOperate.On();
+//Console.WriteLine(operateResult.ToJson().JsonFormatting());
+
+kafkaOperate.OnEvent += KafkaOperate_OnEvent;
+
+operateResult = kafkaOperate.Subscribe("VariableTopic");
+//Console.WriteLine(operateResult.ToJson().JsonFormatting());
+
+while (true)
 {
-    AddressName = "DB71.DBD4",
-    AddressDataType = YSAI.Core.@enum.DataType.Float,
-    AddressType = AddressType.Reality
-});
-OperateResult operateResult = s7ClientOperate.On();
-Console.WriteLine(operateResult.ToJson());
+    operateResult = kafkaOperate.Produce("VariableTopic", new Random().NextDouble().ToString());
+    //Console.WriteLine(operateResult.ToJson().JsonFormatting());
+}
+void KafkaOperate_OnEvent(object? sender, EventResult e)
+{
+    Console.WriteLine(e.ToJson().JsonFormatting());
+}
 
 
-//读取
-operateResult = s7ClientOperate.Read(address);
-Console.WriteLine(operateResult.ToJson());
 
+//S7ClientOperate s7ClientOperate = S7ClientOperate.Instance(new S7ClientData.Basics()
+//{
+//    Ip = "192.168.2.20",
+//    Port = 102,
+//    Rack = 0,
+//    Slot = 1,
+//    S7CpuType = CpuType.S71200,
+//    SN = "1",
+//    DataChangeOut=true
+//});
 
+////点位地址
+//Address address = new Address();
+//address.SN = Guid.NewGuid().ToString();
+//address.CreationTime = DateTime.Now.ToLocalTime();
+//address.AddressArray = new List<AddressDetails>();
+//address.AddressArray.Add(new AddressDetails()
+//{
+//    AddressName = "DB71.DBD4",
+//    AddressDataType = YSAI.Core.@enum.DataType.Float,
+//    AddressType = AddressType.Reality
+//});
+//OperateResult operateResult = s7ClientOperate.On();
+//Console.WriteLine(operateResult.ToJson());
 
-s7ClientOperate.OnEvent += S7ClientOperate_OnEvent;
 
+////读取
+//operateResult = s7ClientOperate.Read(address);
+//Console.WriteLine(operateResult.ToJson());
 
-operateResult = s7ClientOperate.Subscribe(address);
-Console.WriteLine(operateResult.ToJson());
 
-while (true)
-{
-    string a = Console.ReadLine();
-    ConcurrentDictionary<string, float> keyValuePairs = new ConcurrentDictionary<string, float>();
-    keyValuePairs.TryAdd("DB71.DBD4", float.Parse(a));
-    operateResult = s7ClientOperate.Write(keyValuePairs);
-    Console.WriteLine(operateResult.ToJson());
-}
-void S7ClientOperate_OnEvent(object? sender, EventResult e)
-{
-    Console.WriteLine(e.ToJson());
-}
+
+//s7ClientOperate.OnEvent += S7ClientOperate_OnEvent;
+
+
+//operateResult = s7ClientOperate.Subscribe(address);
+//Console.WriteLine(operateResult.ToJson());
+
+//while (true)
+//{
+//    string a = Console.ReadLine();
+//    ConcurrentDictionary<string, float> keyValuePairs = new ConcurrentDictionary<string, float>();
+//    keyValuePairs.TryAdd("DB71.DBD4", float.Parse(a));
+//    operateResult = s7ClientOperate.Write(keyValuePairs);
+//    Console.WriteLine(operateResult.ToJson());
+//}
+//void S7ClientOperate_OnEvent(object? sender, EventResult e)
+//{
+//    Console.WriteLine(e.ToJson());
+//}
 
 
 //RabbitMQOperate rabbitMQOperate = RabbitMQOperate.Instance(new RabbitMQData.Basics()

src/YSAI.DAQ/YSAI.Unility/YSAI.Unility.csproj

@@ -13,7 +13,7 @@
 	</PropertyGroup>
 
 	<ItemGroup>
-		<PackageReference Include="Microsoft.IdentityModel.Tokens" Version="6.32.2" />
+		<PackageReference Include="Microsoft.IdentityModel.Tokens" Version="6.32.3" />
 		<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
 	</ItemGroup>