UdpServer.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using NewLife.Data;
using NewLife.Model;

namespace NewLife.Net
{
    /// <summary>增强的UDP</summary>
    /// <remarks>
    /// 如果已经打开异步接收，还要使用同步接收，则同步Receive内部不再调用底层Socket，而是等待截走异步数据。
    /// </remarks>
    public class UdpServer : SessionBase, ISocketServer
    {
        #region 属性
        /// <summary>会话超时时间。默认30秒</summary>
        /// <remarks>
        /// 对于每一个会话连接，如果超过该时间仍然没有收到任何数据，则断开会话连接。
        /// </remarks>
        public Int32 SessionTimeout { get; set; }

        /// <summary>最后一次同步接收数据得到的远程地址</summary>
        public IPEndPoint LastRemote { get; set; }

        /// <summary>是否接收来自自己广播的环回数据。默认false</summary>
        public Boolean Loopback { get; set; }

        /// <summary>会话统计</summary>
        public IStatistics StatSession { get; set; }

        /// <summary>粘包处理接口</summary>
        public IPacketFactory SessionPacket { get; set; }
        #endregion

        #region 构造
        /// <summary>实例化增强UDP</summary>
        public UdpServer()
        {
            Local.Type = NetType.Udp;
            Remote.Type = NetType.Udp;
            _Sessions = new SessionCollection(this);

            StatSession = new Statistics();

            SessionTimeout = Setting.Current.SessionTimeout;
        }

        /// <summary>使用监听口初始化</summary>
        /// <param name="listenPort"></param>
        public UdpServer(Int32 listenPort)
            : this()
        {
            Port = listenPort;
        }
        #endregion

        #region 方法
        /// <summary>打开</summary>
        protected override Boolean OnOpen()
        {
            if (Client == null || !Client.IsBound)
            {
                // 根据目标地址适配本地IPv4/IPv6
                if (Remote != null && !Remote.Address.IsAny())
                {
                    Local.Address = Local.Address.GetRightAny(Remote.Address.AddressFamily);
                }

                Client = NetHelper.CreateUdp(Local.EndPoint.Address.IsIPv4());
                Client.Bind(Local.EndPoint);
                CheckDynamic();

                WriteLog("Open {0}", this);
            }

            return true;
        }

        /// <summary>关闭</summary>
        protected override Boolean OnClose(String reason)
        {
            if (Client != null)
            {
                WriteLog("Close {0} {1}", reason, this);

                var udp = Client;
                Client = null;
                try
                {
                    CloseAllSession();

                    udp.Shutdown();
                }
                catch (Exception ex)
                {
                    if (!ex.IsDisposed()) OnError("Close", ex);
                    if (ThrowException) throw;

                    return false;
                }
            }

            return true;
        }
        #endregion

        #region 发送
        /// <summary>发送数据</summary>
        /// <remarks>
        /// 目标地址由<seealso cref="SessionBase.Remote"/>决定
        /// </remarks>
        /// <param name="pk">数据包</param>
        /// <returns>是否成功</returns>
        protected override Boolean OnSend(Packet pk)
        {
            var count = pk.Total;

            StatSend?.Increment(count);

            try
            {
                var sp = Client;
                lock (sp)
                {
                    if (Client.Connected)
                    {
                        if (Log.Enable && LogSend) WriteLog("Send [{0}]: {1}", count, pk.ToHex());

                        if (pk.Next == null)
                            sp.Send(pk.Data, pk.Offset, count, SocketFlags.None);
                        else
                            sp.Send(pk.ToArray(), 0, count, SocketFlags.None);
                    }
                    else
                    {
                        Client.CheckBroadcast(Remote.Address);
                        if (Log.Enable && LogSend) WriteLog("Send {2} [{0}]: {1}", count, pk.ToHex(), Remote.EndPoint);

                        if (pk.Next == null)
                            sp.SendTo(pk.Data, pk.Offset, count, SocketFlags.None, Remote.EndPoint);
                        else
                            sp.SendTo(pk.ToArray(), 0, count, SocketFlags.None, Remote.EndPoint);
                    }
                }

                return true;
            }
            catch (Exception ex)
            {
                if (!ex.IsDisposed())
                {
                    OnError("Send", ex);

                    // 发送异常可能是连接出了问题，UDP不需要关闭
                    //Close();

                    if (ThrowException) throw;
                }
                return false;
            }
        }

        /// <summary>发送数据包到目的地址</summary>
        /// <param name="pk"></param>
        /// <returns></returns>
        public override async Task<Packet> SendAsync(Packet pk)
        {
            return await SendAsync(pk, Remote.EndPoint, true);
        }

        /// <summary>发送数据包到目的地址</summary>
        /// <param name="pk"></param>
        /// <param name="remote"></param>
        /// <param name="wait"></param>
        /// <returns></returns>
        internal async Task<Packet> SendAsync(Packet pk, IPEndPoint remote, Boolean wait)
        {
            if (pk.Count > 0)
            {
                if (remote != null && remote.Address == IPAddress.Broadcast && !Client.EnableBroadcast)
                {
                    Client.EnableBroadcast = true;
                    // 广播匹配任意响应
                    remote = null;
                }
            }

            if (Packet == null) Packet = new PacketProvider();

            var task = !wait ? null : Packet.Add(pk, remote, Timeout);

            // 这里先发送，基类的SendAsync注定发给Remote而不是remote
            if (!SendByQueue(pk, remote)) return null;

            if (!wait) return null;

            return await task;
        }

        internal override Boolean OnSendAsync(SocketAsyncEventArgs se)
        {
            if (se.RemoteEndPoint == null) se.RemoteEndPoint = Remote.EndPoint;

            return Client.SendToAsync(se);
        }
        #endregion

        #region 接收
        /// <summary>处理收到的数据</summary>
        /// <param name="pk"></param>
        /// <param name="remote"></param>
        protected override Boolean OnReceive(Packet pk, IPEndPoint remote)
        {
            // 过滤自己广播的环回数据。放在这里，兼容UdpSession
            if (!Loopback && remote.Port == Port)
            {
                if (!Local.Address.IsAny())
                {
                    if (remote.Address.Equals(Local.Address)) return false;
                }
                else
                {
                    foreach (var item in NetHelper.GetIPsWithCache())
                    {
                        if (remote.Address.Equals(item)) return false;
                    }
                }
            }

#if !__MOBILE__
            // 更新全局远程IP地址
            NewLife.Web.WebHelper.UserHost = remote?.Address + "";
#endif
            LastRemote = remote;

            StatReceive?.Increment(pk.Count);
            if (base.OnReceive(pk, remote)) return true;

            // 分析处理
            var e = new ReceivedEventArgs(pk)
            {
                UserState = remote
            };

            // 为该连接单独创建一个会话，方便直接通信
            var session = CreateSession(remote);
            // 数据直接转交给会话，不再经过事件，那样在会话较多时极为浪费资源
            if (session is UdpSession us)
                us.OnReceive(e);
            else
            {
                // 没有匹配到任何会话时，才在这里显示日志。理论上不存在这个可能性
                if (Log.Enable && LogReceive) WriteLog("Recv [{0}]: {1}", e.Length, e.ToHex(32, null));
            }

            if (session != null) RaiseReceive(session, e);

            return true;
        }

        /// <summary>收到异常时如何处理。Tcp/Udp客户端默认关闭会话，但是Udp服务端不能关闭服务器，仅关闭会话</summary>
        /// <param name="se"></param>
        /// <returns>是否当作异常处理并结束会话</returns>
        internal override Boolean OnReceiveError(SocketAsyncEventArgs se)
        {
            // 缓冲区不足时，加大
            if (se.SocketError == SocketError.MessageSize && BufferSize < 1024 * 1024) BufferSize *= 2;

            // Udp服务器不能关闭自己，但是要关闭会话
            // Udp客户端一般不关闭自己
            if (se.SocketError != SocketError.ConnectionReset &&
                se.SocketError != SocketError.ConnectionAborted
                ) return true;

            // 以下仅处理Reset
            //if (!EnableReset) return false;

            // 关闭相应会话
            var sessions = _Sessions;
            if (sessions != null)
            {
                var ep = se.RemoteEndPoint as IPEndPoint;
                var ss = sessions.Get(ep + "");
                if (ss != null) ss.Dispose();
            }
            // 无论如何，Udp都不关闭自己
            return false;
        }

        internal override Boolean OnReceiveAsync(SocketAsyncEventArgs se)
        {
            if (!Active || Client == null) return false;

            // 每次接收以后，这个会被设置为远程地址，这里重置一下，以防万一
            se.RemoteEndPoint = new IPEndPoint(IPAddress.Any.GetRightAny(Local.EndPoint.AddressFamily), 0);

            return Client.ReceiveFromAsync(se);
        }
        #endregion

        #region 会话
        /// <summary>新会话时触发</summary>
        public event EventHandler<SessionEventArgs> NewSession;

        private SessionCollection _Sessions;
        /// <summary>会话集合。用地址端口作为标识，业务应用自己维持地址端口与业务主键的对应关系。</summary>
        public IDictionary<String, ISocketSession> Sessions { get { return _Sessions; } }

        Int32 g_ID = 0;
        /// <summary>创建会话</summary>
        /// <param name="remoteEP"></param>
        /// <returns></returns>
        public virtual ISocketSession CreateSession(IPEndPoint remoteEP)
        {
            if (Disposed) throw new ObjectDisposedException(GetType().Name);

            var sessions = _Sessions;
            if (sessions == null) return null;

            // 平均执行耗时260.80ns，其中55%花在sessions.Get上面，Get里面有加锁操作

            if (!Active)
            {
                // 根据目标地址适配本地IPv4/IPv6
                Local.Address = Local.Address.GetRightAny(remoteEP.AddressFamily);

                if (!Open()) return null;
            }

            // 需要查找已有会话，已有会话不存在时才创建新会话
            var session = sessions.Get(remoteEP + "");
            if (session != null) return session;

            // 相同远程地址可能同时发来多个数据包，而底层采取多线程方式同时调度，导致创建多个会话
            lock (sessions)
            {
                // 需要查找已有会话，已有会话不存在时才创建新会话
                session = sessions.Get(remoteEP + "");
                if (session != null) return session;

                var us = new UdpSession(this, remoteEP)
                {
                    Log = Log,
                    LogSend = LogSend,
                    LogReceive = LogReceive,
                    // UDP不好分会话统计
                    //us.StatSend.Parent = StatSend;
                    //us.StatReceive.Parent = StatReceive;
                    Packet = SessionPacket?.Create()
                };

                session = us;
                if (sessions.Add(session))
                {
                    //us.ID = g_ID++;
                    // 会话改为原子操作，避免多线程冲突
                    us.ID = Interlocked.Increment(ref g_ID);
                    us.Start();

                    if (StatSession != null) StatSession.Increment(1);

                    // 触发新会话事件
                    NewSession?.Invoke(this, new SessionEventArgs { Session = session });
                }
            }

            return session;
        }

        private void CloseAllSession()
        {
            var sessions = _Sessions;
            if (sessions != null)
            {
                if (sessions.Count > 0)
                {
                    WriteLog("准备释放会话{0}个！", sessions.Count);
                    sessions.TryDispose();
                    sessions.Clear();
                }
            }
        }
        #endregion

        #region IServer接口
        void IServer.Start() { Open(); }

        void IServer.Stop(String reason) { Close(reason ?? "服务停止"); }
        #endregion

        #region 辅助
        /// <summary>已重载。</summary>
        /// <returns></returns>
        public override String ToString()
        {
            var ss = Sessions;
            if (ss != null && ss.Count > 0)
                return String.Format("{0} [{1}]", Local, ss.Count);
            else
                return Local.ToString();
        }
        #endregion
    }

    /// <summary>Udp扩展</summary>
    public static class UdpHelper
    {
        /// <summary>发送数据流</summary>
        /// <param name="udp"></param>
        /// <param name="stream"></param>
        /// <param name="remoteEP"></param>
        /// <returns>返回自身，用于链式写法</returns>
        public static UdpClient Send(this UdpClient udp, Stream stream, IPEndPoint remoteEP = null)
        {
            Int64 total = 0;

            var size = 1472;
            var buffer = new Byte[size];
            while (true)
            {
                var n = stream.Read(buffer, 0, buffer.Length);
                if (n <= 0) break;

                udp.Send(buffer, n, remoteEP);
                total += n;

                if (n < buffer.Length) break;
            }
            return udp;
        }

        /// <summary>向指定目的地发送信息</summary>
        /// <param name="udp"></param>
        /// <param name="buffer">缓冲区</param>
        /// <param name="remoteEP"></param>
        /// <returns>返回自身，用于链式写法</returns>
        public static UdpClient Send(this UdpClient udp, Byte[] buffer, IPEndPoint remoteEP = null)
        {
            udp.Send(buffer, buffer.Length, remoteEP);
            return udp;
        }

        /// <summary>向指定目的地发送信息</summary>
        /// <param name="udp"></param>
        /// <param name="message"></param>
        /// <param name="encoding">文本编码，默认null表示UTF-8编码</param>
        /// <param name="remoteEP"></param>
        /// <returns>返回自身，用于链式写法</returns>
        public static UdpClient Send(this UdpClient udp, String message, Encoding encoding = null, IPEndPoint remoteEP = null)
        {
            if (encoding == null)
                Send(udp, Encoding.UTF8.GetBytes(message), remoteEP);
            else
                Send(udp, encoding.GetBytes(message), remoteEP);
            return udp;
        }

        /// <summary>广播数据包</summary>
        /// <param name="udp"></param>
        /// <param name="buffer">缓冲区</param>
        /// <param name="port"></param>
        public static UdpClient Broadcast(this UdpClient udp, Byte[] buffer, Int32 port)
        {
            if (udp.Client != null && udp.Client.LocalEndPoint != null)
            {
                var ip = udp.Client.LocalEndPoint as IPEndPoint;
                if (!ip.Address.IsIPv4()) throw new NotSupportedException("IPv6不支持广播！");
            }

            if (!udp.EnableBroadcast) udp.EnableBroadcast = true;

            udp.Send(buffer, buffer.Length, new IPEndPoint(IPAddress.Broadcast, port));

            return udp;
        }

        /// <summary>广播字符串</summary>
        /// <param name="udp"></param>
        /// <param name="message"></param>
        /// <param name="port"></param>
        public static UdpClient Broadcast(this UdpClient udp, String message, Int32 port)
        {
            var buffer = Encoding.UTF8.GetBytes(message);
            return Broadcast(udp, buffer, port);
        }

        /// <summary>接收字符串</summary>
        /// <param name="udp"></param>
        /// <param name="encoding">文本编码，默认null表示UTF-8编码</param>
        /// <returns></returns>
        public static String ReceiveString(this UdpClient udp, Encoding encoding = null)
        {
            IPEndPoint ep = null;
            var buffer = udp.Receive(ref ep);
            if (buffer == null || buffer.Length < 1) return null;

            if (encoding == null) encoding = Encoding.UTF8;
            return encoding.GetString(buffer);
        }
    }
}