using System.Collections.Concurrent; using System.Net; using System.Net.Sockets; using System.Runtime.CompilerServices; using System.Text; using DotNext; using Newtonsoft.Json; using WebProtocol; /// UDP接受数据包格式 public class UDPData { /// /// 接受到的时间 /// public required DateTime DateTime { get; set; } /// /// 发送来源的IP地址 /// public required string Address { get; set; } /// /// 发送来源的端口号 /// public required int Port { get; set; } /// /// 任务ID /// public required int TaskID { get; set; } /// /// 接受到的数据 /// public required byte[] Data { get; set; } /// /// 是否被读取过 /// public required bool HasRead { get; set; } /// /// 深度拷贝对象 /// /// UDPData public UDPData DeepClone() { var cloneData = new byte[this.Data.Length]; Buffer.BlockCopy(this.Data, 0, cloneData, 0, this.Data.Length); return new UDPData() { DateTime = this.DateTime, Address = new string(this.Address), Port = this.Port, TaskID = this.TaskID, Data = cloneData, HasRead = this.HasRead }; } /// /// 将UDP Data 转化为Json 格式字符串 /// /// json字符串 public override string ToString() { return JsonConvert.SerializeObject(this); } } /// /// UDP 服务器 /// public class UDPServer { private static NLog.Logger logger = NLog.LogManager.GetCurrentClassLogger(); private ConcurrentDictionary> udpData = new ConcurrentDictionary>(); private int listenPort; private List listeners = new List(); private IPEndPoint groupEP; private bool isRunning = false; /// /// 是否正在工作 /// public bool IsRunning { get { return isRunning; } } /// UDP 服务器的错误代码 public enum ErrorCode { /// [TODO:description] Success = 0, /// [TODO:description] GetNoneAfterTimeout, /// [TODO:description] ResponseWrong, /// [TODO:description] NotRecvDataPackage, } /// /// Construct a udp server with fixed port /// /// Device UDP Port /// UDP Client Num /// UDPServer class public UDPServer(int port, int num) { // Construction this.listenPort = port; try { for (int i = 0; i < num; i++) { listeners.Add(new UdpClient(this.listenPort + i)); } this.groupEP = new IPEndPoint(IPAddress.Any, listenPort); } catch (Exception e) { Console.WriteLine(e.ToString()); throw new ArgumentException( $"Failed to set up server with this port: {port}", nameof(port) ); } } /// /// 异步寻找目标发送的内容 /// /// 目标IP地址 /// [TODO:parameter] /// 超时时间 /// 延迟时间 /// 调用函数名称 /// 调用函数位置 /// /// 异步Optional 数据包: /// Optional 为空时,表明找不到数据; /// Optional 存在时,为最先收到的数据 /// public async ValueTask> FindDataAsync( string ipAddr, int taskID, int timeout = 1000, int cycle = 0, [CallerMemberName] string callerName = "", [CallerLineNumber] int callerLineNum = 0 ) { UDPData? data = null; var startTime = DateTime.Now; var isTimeout = false; while (!isTimeout) { var elapsed = DateTime.Now - startTime; isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout); if (isTimeout) break; lock (udpData) { if (udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue) && dataQueue.TryDequeue(out data)) { // logger.Debug($"Find UDP Data: {data.ToString()}"); break; } } await Task.Delay(cycle); } if (data is null) { logger.Trace("Get nothing even after time out"); return new(null); } else { return new(data.DeepClone()); } } /// /// 异步寻找目标发送的所有内容,并清空队列 /// /// 目标IP地址 /// 任务ID /// 超时时间 /// 异步Optional 数据包列表 public async ValueTask>> FindDataArrayAsync(string ipAddr, int taskID, int timeout = 1000) { List? data = null; var startTime = DateTime.Now; var isTimeout = false; while (!isTimeout) { var elapsed = DateTime.Now - startTime; isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout); if (isTimeout) break; lock (udpData) { if (udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue) && !dataQueue.IsEmpty) { data = new List(); while (dataQueue.TryDequeue(out var item)) { data.Add(item); } break; } } } if (data is null) { logger.Trace("Get nothing even after time out"); return new(null); } else { return new(data); } } /// /// 获取还未被读取的数据列表 /// /// IP地址 /// 任务ID /// 超时时间 /// 数据列表 public async ValueTask>> GetDataArrayAsync(string ipAddr, int taskID, int timeout = 1000) { List? data = null; var startTime = DateTime.Now; var isTimeout = false; while (!isTimeout) { var elapsed = DateTime.Now - startTime; isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout); if (isTimeout) break; if (udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue) && !dataQueue.IsEmpty) { data = dataQueue.ToArray().ToList(); // logger.Debug($"Find UDP Data Array: {JsonConvert.SerializeObject(data)}"); break; } } if (data is null) { logger.Trace("Get nothing even after time out"); return new(null); } else { return new(data); } } /// /// 异步获取指定IP和任务ID的数据队列长度 /// /// IP地址 /// 任务ID /// 超时时间 /// 数据队列长度 public async ValueTask> GetDataCountAsync(string ipAddr, int taskID, int timeout = 1000) { int? count = null; var startTime = DateTime.Now; var isTimeout = false; while (!isTimeout) { var elapsed = DateTime.Now - startTime; isTimeout = elapsed >= TimeSpan.FromMilliseconds(timeout); if (isTimeout) break; if (udpData.TryGetValue($"{ipAddr}-{taskID}", out var dataQueue)) { count = dataQueue.Count; break; } } if (count is null) { logger.Trace("Get nothing even after time out"); return Optional.None; } else { return new(count.Value); } } /// /// 异步等待写响应 /// /// IP地址 /// [TODO:parameter] /// UDP 端口 /// 超时时间范围 /// 接收响应包 public async ValueTask> WaitForAckAsync (string address, int taskID, int port = -1, int timeout = 1000) { var data = await FindDataAsync(address, taskID, timeout); if (!data.HasValue) return new(new Exception("Get None even after time out!")); var recvData = data.Value; if (recvData.Address != address || (port > 0 && recvData.Port != port)) return new(new Exception("Receive Data From Wrong Board!")); var retPack = WebProtocol.RecvRespPackage.FromBytes(recvData.Data); if (!retPack.IsSuccessful) return new(new Exception("Not RecvDataPackage!", retPack.Error)); return retPack.Value; } /// /// 异步等待数据 /// /// IP地址 /// [TODO:parameter] /// UDP 端口 /// 超时时间范围 /// 接收数据包 public async ValueTask> WaitForDataAsync (string address, int taskID, int port = -1, int timeout = 1000) { var data = await FindDataAsync(address, taskID, timeout); if (!data.HasValue) return new(new Exception("Get None even after time out!")); var recvData = data.Value; if (recvData.Address != address || (port >= 0 && recvData.Port != port)) return new(new Exception("Receive Data From Wrong Board!")); var retPack = WebProtocol.RecvDataPackage.FromBytes(recvData.Data); if (!retPack.IsSuccessful) return new(new Exception("Not RecvDataPackage!", retPack.Error)); return retPack.Value; } private async Task ReceiveHandler(byte[] data, IPEndPoint endPoint, DateTime time) { // 异步锁保护 udpData await Task.Run(() => { // Handle RemoteEP if (endPoint is null) { logger.Debug($"Receive Data from Unknown at {DateTime.Now.ToString()}:"); logger.Debug($" Original Data : {BitConverter.ToString(data).Replace("-", " ")}"); return; } var udpDataObj = RecordUDPData(data, endPoint, time, Convert.ToInt32(data[1])); // PrintData(udpDataObj); }); } private UDPData RecordUDPData(byte[] bytes, IPEndPoint remoteEP, DateTime time, int taskID) { var remoteAddress = remoteEP.Address.ToString(); var remotePort = remoteEP.Port; var data = new UDPData() { Address = remoteAddress, Port = remotePort, TaskID = taskID, Data = bytes, DateTime = time, HasRead = false, }; lock (udpData) { var key = $"{remoteAddress}-{taskID}"; var dataQueue = udpData.GetOrAdd(key, _ => new ConcurrentQueue()); dataQueue.Enqueue(data); // 对队列进行一次按时间排序 if (dataQueue.Count > 0) { var sorted = dataQueue.OrderBy(d => d.DateTime).ToList(); udpData.TryUpdate(key, new ConcurrentQueue(sorted), dataQueue); } PrintAllData(); } return data; } /// /// 输出UDP Data到log中 /// /// UDP数据 public string PrintData(UDPData data) { var bytes = data.Data; var sign = bytes[0]; string recvData = ""; if (sign == (byte)WebProtocol.PackSign.SendAddr) { var resData = WebProtocol.SendAddrPackage.FromBytes(bytes); if (resData.IsSuccessful) recvData = resData.Value.ToString(); else recvData = resData.Error.ToString(); } else if (sign == (byte)WebProtocol.PackSign.SendData) { } else if (sign == (byte)WebProtocol.PackSign.RecvData) { var resData = WebProtocol.RecvDataPackage.FromBytes(bytes); if (resData.IsSuccessful) recvData = resData.Value.Options.ToString(); else recvData = resData.Error.ToString(); } else if (sign == (byte)WebProtocol.PackSign.RecvResp) { var resData = WebProtocol.RecvRespPackage.FromBytes(bytes); if (resData.IsSuccessful) recvData = resData.Value.Options.ToString(); else recvData = resData.Error.ToString(); } else { recvData = Encoding.ASCII.GetString(bytes, 0, bytes.Length); } // logger.Debug($"Receive Data from {data.Address}:{data.Port} at {data.DateTime.ToString()}:"); // logger.Debug($" Original Data : {BitConverter.ToString(bytes).Replace("-", " ")}"); // if (recvData.Length != 0) logger.Debug($" Decoded Data : {recvData}"); return $@" Receive Data from {data.Address}:{data.Port} at {data.DateTime.ToString()}: Original Data : {BitConverter.ToString(bytes).Replace("-", " ")} Decoded Data : {recvData} "; } /// /// 将所有数据输出到log中 /// /// void public void PrintAllData() { logger.Debug("Ready Data:"); foreach (var kvp in udpData) { foreach (var data in kvp.Value) { logger.Debug(PrintData(data)); } } } /// /// 清空指定IP地址的数据 /// /// IP地址 /// [TODO:parameter] /// public void ClearUDPData(string ipAddr, int taskID) { var key = $"{ipAddr}-{taskID}"; if (udpData.TryGetValue(key, out var dataQueue)) { // 清空队列的最有效方式是替换为新的队列 udpData.TryUpdate(key, new ConcurrentQueue(), dataQueue); } } /// /// Start UDP Server /// /// None public void Start() { this.isRunning = true; try { foreach (var client in listeners) { Task.Run(async () => { while (this.isRunning) { try { UdpReceiveResult result = await client.ReceiveAsync(); ReceiveHandler(result.Buffer, result.RemoteEndPoint, DateTime.Now); } catch (Exception ex) { Console.WriteLine($"Error: {ex.Message}"); } } }); } } catch (Exception e) { Console.WriteLine(e.ToString()); this.isRunning = false; } } /// /// Close UDP Server /// /// None public void Stop() { foreach (var item in listeners) { item.Close(); } this.isRunning = false; } }