Add ChannelAuchenticateTimeout Event.

aaasoft · aaasoft · commit 9a1206fd9b3c · 2021-12-17T10:56:28.000+08:00
diff --git a/Quick.Protocol.WebSocket.Server.AspNetCore/QpWebSocketServer.cs b/Quick.Protocol.WebSocket.Server.AspNetCore/QpWebSocketServer.cs
@@ -1,5 +1,4 @@
 ﻿using Microsoft.AspNetCore.Http;
-using Microsoft.Extensions.Logging;
 using Quick.Protocol.Utils;
 using System;
 using System.Collections;
@@ -12,7 +11,7 @@ namespace Quick.Protocol.WebSocket.Server.AspNetCore
     public class QpWebSocketServer : QpServer
     {
         private Queue<WebSocketContext> webSocketContextQueue = new Queue<WebSocketContext>();
-        private AutoResetEvent waitForConnectionAutoResetEvent;
+        private bool isStarted = false;
 
         private class WebSocketContext
         {
@@ -32,14 +31,18 @@ public QpWebSocketServer(QpWebSocketServerOptions options) : base(options) { }
 
         public override void Start()
         {
-            waitForConnectionAutoResetEvent = new AutoResetEvent(false);
+            isStarted=true;
             lock (webSocketContextQueue)
                 webSocketContextQueue.Clear();
             base.Start();
         }
 
         public Task OnNewConnection(System.Net.WebSockets.WebSocket webSocket, ConnectionInfo connectionInfo)
         {
+            //如果还没有开始接收，则直接关闭
+            if (!isStarted)
+                return webSocket.CloseAsync(System.Net.WebSockets.WebSocketCloseStatus.NormalClosure, "", CancellationToken.None);
+
             var connectionInfoStr = $"WebSocket:{connectionInfo.RemoteIpAddress}:{connectionInfo.RemotePort}";
             var cts = new CancellationTokenSource();
             lock (webSocketContextQueue)
@@ -48,7 +51,6 @@ public Task OnNewConnection(System.Net.WebSockets.WebSocket webSocket, Connectio
                         connectionInfoStr,
                         webSocket,
                         cts));
-            waitForConnectionAutoResetEvent.Set();
             return Task.Delay(-1, cts.Token).ContinueWith(t =>
              {
                  if (LogUtils.LogConnection)
@@ -58,40 +60,42 @@ public Task OnNewConnection(System.Net.WebSockets.WebSocket webSocket, Connectio
 
         public override void Stop()
         {
+            isStarted=false;
             lock (webSocketContextQueue)
                 webSocketContextQueue.Clear();
-            waitForConnectionAutoResetEvent?.Dispose();
             base.Stop();
         }
 
-        protected override Task InnerAcceptAsync(CancellationToken token)
+        protected override async Task InnerAcceptAsync(CancellationToken token)
         {
-            return Task.Run(() =>
+            WebSocketContext[] webSocketContexts = null;
+            lock (webSocketContextQueue)
+            {
+                webSocketContexts = webSocketContextQueue.ToArray();
+                webSocketContextQueue.Clear();
+            }
+            //如果当前没有WebSocket连接，则等待0.1秒后再返回
+            if (webSocketContexts == null || webSocketContexts.Length==0)
+            {
+                await Task.Delay(100);
+                return;
+            }
+            foreach (var context in webSocketContexts)
             {
-                waitForConnectionAutoResetEvent.WaitOne();
-                WebSocketContext[] webSocketContexts = null;
-                lock (webSocketContextQueue)
+                try
                 {
-                    webSocketContexts = webSocketContextQueue.ToArray();
-                    webSocketContextQueue.Clear();
+                    if (LogUtils.LogConnection)
+                        Console.WriteLine("[Connection]{0} connected.", context.ConnectionInfo);
+                    OnNewChannelConnected(new WebSocketServerStream(context.WebSocket, context.Cts), context.ConnectionInfo, token);
                 }
-                foreach (var context in webSocketContexts)
+                catch (Exception ex)
                 {
-                    try
-                    {
-                        if (LogUtils.LogConnection)
-                            Console.WriteLine("[Connection]{0} connected.", context.ConnectionInfo);
-                        OnNewChannelConnected(new WebSocketServerStream(context.WebSocket,context.Cts), context.ConnectionInfo, token);
-                    }
-                    catch (Exception ex)
-                    {
-                        if (LogUtils.LogConnection)
-                            Console.WriteLine("[Connection]Init&Start Channel error,reason:{0}", ex.ToString());
-                        try { context.WebSocket.CloseAsync(System.Net.WebSockets.WebSocketCloseStatus.InternalServerError, ex.Message, CancellationToken.None); }
-                        catch { }
-                    }
+                    if (LogUtils.LogConnection)
+                        Console.WriteLine("[Connection]Init&Start Channel error,reason:{0}", ex.ToString());
+                    try { await context.WebSocket.CloseAsync(System.Net.WebSockets.WebSocketCloseStatus.InternalServerError, ex.Message, CancellationToken.None); }
+                    catch { }
                 }
-            });
+            }
         }
     }
 }
diff --git a/Quick.Protocol/QpChannel.cs b/Quick.Protocol/QpChannel.cs
@@ -57,10 +57,30 @@ public abstract class QpChannel
         private ConcurrentDictionary<string, CommandContext> commandDict = new ConcurrentDictionary<string, CommandContext>();
 
         /// <summary>
-        /// 当时是否连接
+        /// 当前是否连接，要连接且认证通过后，才设置此属性为true
         /// </summary>
         public bool IsConnected { get; protected set; }
 
+        /// <summary>
+        /// 连接断开时
+        /// </summary>
+        public event EventHandler Disconnected;
+
+        /// <summary>
+        /// 断开连接时
+        /// </summary>
+        protected virtual void Disconnect()
+        {
+            lock (this)
+            {
+                if (IsConnected)
+                {
+                    IsConnected = false;
+                    Disconnected?.Invoke(this, QpEventArgs.Empty);
+                }
+            }
+        }
+
         /// <summary>
         /// 最后的异常
         /// </summary>
@@ -182,6 +202,7 @@ protected virtual void OnReadError(Exception exception)
             LastException = exception;
             LogUtils.Log("[ReadError]{0}: {1}", DateTime.Now, ExceptionUtils.GetExceptionString(exception));
             InitQpPackageHandler_Stream(null);
+            Disconnect();
         }
 
         //获取空闲的缓存
diff --git a/Quick.Protocol/QpClient.cs b/Quick.Protocol/QpClient.cs
@@ -12,12 +12,7 @@ namespace Quick.Protocol
     public abstract class QpClient : QpChannel
     {
         private CancellationTokenSource cts = null;
-        public QpClientOptions Options { get; private set; }        
-
-        /// <summary>
-        /// 连接断开时
-        /// </summary>
-        public event EventHandler Disconnected;
+        public QpClientOptions Options { get; private set; }
 
         public QpClient(QpClientOptions options)
             : base(options)
@@ -39,8 +34,6 @@ public async Task ConnectAsync()
             var token = cts.Token;
 
             var stream = await InnerConnectAsync();
-            IsConnected = true;
-
             //初始化网络
             InitQpPackageHandler_Stream(stream);
 
@@ -69,6 +62,7 @@ public async Task ConnectAsync()
             }, 5000, () =>
             {
                 Options.OnAuthPassed();
+                IsConnected=true;
             });
 
             //开始心跳
@@ -96,23 +90,15 @@ private void cancellAll()
             }
         }
 
-        protected virtual void Disconnect()
-        {
-            if (IsConnected)
-            {
-                IsConnected = false;
-                Disconnected?.Invoke(this, QpEventArgs.Empty);                
-            }
-        }
-
         /// <summary>
         /// 关闭连接
         /// </summary>
         public void Close()
         {
             cancellAll();
+            IsConnected = false;
             InitQpPackageHandler_Stream(null);
-            Disconnect();            
+            Disconnect();
         }
     }
 }
diff --git a/Quick.Protocol/QpServer.cs b/Quick.Protocol/QpServer.cs
@@ -12,23 +12,18 @@ public abstract class QpServer
     {
         private CancellationTokenSource cts;
         private QpServerOptions options;
+        
         private List<QpServerChannel> channelList = new List<QpServerChannel>();
-        private List<QpServerChannel> auchenticatedChannelList = new List<QpServerChannel>();
 
         /// <summary>
         /// 增加Tag属性，用于引用与QpServer相关的对象
         /// </summary>
         public Object Tag { get; set; }
 
-        /// <summary>
-        /// 获取全部的通道
-        /// </summary>
-        public QpServerChannel[] Channels { get; private set; } = new QpServerChannel[0];
-
         /// <summary>
         /// 已通过认证的通道
         /// </summary>
-        public QpServerChannel[] AuchenticatedChannels { get; private set; } = new QpServerChannel[0];
+        public QpServerChannel[] Channels { get; private set; } = new QpServerChannel[0];
 
         /// <summary>
         /// 通道连接上时
@@ -39,6 +34,11 @@ public abstract class QpServer
         /// 通道连接断开时
         /// </summary>
         public event EventHandler<QpServerChannel> ChannelDisconnected;
+        
+        /// <summary>
+        /// 通道认证超时
+        /// </summary>
+        public event EventHandler<QpServerChannel> ChannelAuchenticateTimeout;
 
         public QpServer(QpServerOptions options)
         {
@@ -60,46 +60,39 @@ internal void RemoveChannel(QpServerChannel channel)
                     channelList.Remove(channel);
                     Channels = channelList.ToArray();
                 }
-            lock (auchenticatedChannelList)
-                if (auchenticatedChannelList.Contains(channel))
-                {
-                    auchenticatedChannelList.Remove(channel);
-                    AuchenticatedChannels = auchenticatedChannelList.ToArray();
-                }
         }
 
         protected void OnNewChannelConnected(Stream stream, string channelName, CancellationToken token)
         {
             var channel = new QpServerChannel(this, stream, channelName, token, options.Clone());
-            //将通道加入到全部通道列表里面
-            lock (channelList)
+
+            //认证超时
+            channel.AuchenticateTimeout += (sender, e) =>
             {
-                channelList.Add(channel);
-                Channels = channelList.ToArray();
-            }
+                if (LogUtils.LogConnection)
+                    LogUtils.Log("[Connection]{0} auchenticate timeout.", channelName);
+                ChannelAuchenticateTimeout?.Invoke(this, channel);
+            };
 
-            //认证通过后，才将通道添加到已认证通道列表里面
+            //认证通过后，才将通道添加到已连接通道列表里面
             channel.Auchenticated += (sender, e) =>
             {
-                lock (auchenticatedChannelList)
+                lock (channelList)
                 {
-                    auchenticatedChannelList.Add(channel);
-                    AuchenticatedChannels = auchenticatedChannelList.ToArray();
+                    channelList.Add(channel);
+                    Channels = channelList.ToArray();
                 }
-            };
-            channel.Disconnected += (sender, e) =>
-            {
-                if (LogUtils.LogConnection)
-                    LogUtils.Log("[Connection]{0} Disconnected.", channelName);
-                RemoveChannel(channel);
-                try { stream.Dispose(); }
-                catch { }
-                ChannelDisconnected?.Invoke(this, channel);
-            };
-            Task.Run(() =>
-            {
                 ChannelConnected?.Invoke(this, channel);
-            });
+                channel.Disconnected += (sender2, e2) =>
+                {
+                    if (LogUtils.LogConnection)
+                        LogUtils.Log("[Connection]{0} Disconnected.", channelName);
+                    RemoveChannel(channel);
+                    try { stream.Dispose(); }
+                    catch { }
+                    ChannelDisconnected?.Invoke(this, channel);
+                };
+            };
         }
 
         protected abstract Task InnerAcceptAsync(CancellationToken token);
diff --git a/Quick.Protocol/QpServerChannel.cs b/Quick.Protocol/QpServerChannel.cs
@@ -25,12 +25,11 @@ public class QpServerChannel : QpChannel
         /// <summary>
         /// 通过认证时
         /// </summary>
-        public event EventHandler Auchenticated;
-
+        internal event EventHandler Auchenticated;
         /// <summary>
-        /// 连接断开时
+        /// 认证超时
         /// </summary>
-        public event EventHandler Disconnected;
+        internal event EventHandler AuchenticateTimeout;
 
         public QpServerChannel(QpServer server, Stream stream, string channelName, CancellationToken cancellationToken, QpServerOptions options) : base(options)
         {
@@ -43,7 +42,6 @@ public QpServerChannel(QpServer server, Stream stream, string channelName, Cance
             cancellationToken.Register(() => Stop());
             //修改缓存大小
             ChangeBufferSize(options.BufferSize);
-            IsConnected = true;
 
             //初始化连接相关指令处理器
             var connectAndAuthCommandExecuterManager = new CommandExecuterManager();
@@ -56,6 +54,25 @@ public QpServerChannel(QpServer server, Stream stream, string channelName, Cance
             InitQpPackageHandler_Stream(stream);
             //开始读取其他数据包
             BeginReadPackage(cts.Token);
+
+            //如果认证超时时间后没有通过认证，则断开连接
+            if (options.AuthenticateTimeout>0)
+                Task.Delay(options.AuthenticateTimeout, cts.Token).ContinueWith(t =>
+                {
+                    if (t.IsCanceled)
+                        return;
+                    if (stream!=null)
+                    {
+                        try
+                        {
+                            stream.Close();
+                            stream.Dispose();
+                            stream=null;
+                        }
+                        catch { }
+                    }
+                    AuchenticateTimeout?.Invoke(this, EventArgs.Empty);
+                });
         }
 
         private Commands.Connect.Response connect(QpChannel handler, Commands.Connect.Request request)
@@ -87,6 +104,7 @@ private Commands.Authenticate.Response authenticate(QpChannel handler, Commands.
                 });
                 throw new CommandException(1, "认证失败！");
             }
+            IsConnected=true;
             Auchenticated?.Invoke(this, EventArgs.Empty);
             return new Commands.Authenticate.Response();
         }
@@ -147,11 +165,6 @@ protected override void OnReadError(Exception exception)
             }
             Stop();
             base.OnReadError(exception);
-            if (IsConnected)
-            {
-                IsConnected = false;
-                Disconnected?.Invoke(this, QpEventArgs.Empty);
-            }
         }
     }
 }
diff --git a/Quick.Protocol/QpServerOptions.cs b/Quick.Protocol/QpServerOptions.cs
@@ -12,7 +12,10 @@ public class QpServerOptions : QpChannelOptions
         /// 缓存大小(默认128KB)
         /// </summary>
         public int BufferSize = 128 * 1024;
-
+        /// <summary>
+        /// 认证超时时间，在指定的超时时间没有完成认证，则断开连接
+        /// </summary>
+        public int AuthenticateTimeout { get; set; } = 5000;
         /// <summary>
         /// 服务端程序
         /// </summary>
