From 2e0260b4c6ae76cbe2b1616e01edcc642cfe1c0e Mon Sep 17 00:00:00 2001 From: wmpadmin Date: Sat, 28 Apr 2018 10:25:34 +0200 Subject: [PATCH 1/4] lister-fix initial --- src/Fleck/Interfaces/ISocket.cs | 2 +- src/Fleck/SocketWrapper.cs | 8 ++++---- src/Fleck/WebSocketServer.cs | 8 ++++++-- 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/src/Fleck/Interfaces/ISocket.cs b/src/Fleck/Interfaces/ISocket.cs index b6481a0a..2a553b19 100644 --- a/src/Fleck/Interfaces/ISocket.cs +++ b/src/Fleck/Interfaces/ISocket.cs @@ -17,7 +17,7 @@ public interface ISocket bool NoDelay { get; set; } EndPoint LocalEndPoint { get; } - Task Accept(Action callback, Action error); + Task Accept(Action callback, Action error); Task Send(byte[] buffer, Action callback, Action error); Task Receive(byte[] buffer, Action callback, Action error, int offset = 0); Task Authenticate(X509Certificate2 certificate, SslProtocols enabledSslProtocols, Action callback, Action error); diff --git a/src/Fleck/SocketWrapper.cs b/src/Fleck/SocketWrapper.cs index eb8a2894..bee73650 100644 --- a/src/Fleck/SocketWrapper.cs +++ b/src/Fleck/SocketWrapper.cs @@ -130,14 +130,14 @@ public Task Receive(byte[] buffer, Action callback, Action } } - public Task Accept(Action callback, Action error) + public Task Accept(Action callback, Action error) { Func end = r => _tokenSource.Token.IsCancellationRequested ? null : new SocketWrapper(_socket.EndAccept(r)); var task = _taskFactory.FromAsync(_socket.BeginAccept, end, null); - task.ContinueWith(t => callback(t.Result), TaskContinuationOptions.OnlyOnRanToCompletion) - .ContinueWith(t => error(t.Exception), TaskContinuationOptions.OnlyOnFaulted); + task.ContinueWith(t => error(t.Exception), TaskContinuationOptions.OnlyOnFaulted); - return task; + + return task.ContinueWith (t => callback (t.Result), TaskContinuationOptions.OnlyOnRanToCompletion); } public void Dispose() diff --git a/src/Fleck/WebSocketServer.cs b/src/Fleck/WebSocketServer.cs index c5ce5721..75bf48ac 100644 --- a/src/Fleck/WebSocketServer.cs +++ b/src/Fleck/WebSocketServer.cs @@ -4,6 +4,7 @@ using System.Security.Cryptography.X509Certificates; using System.Collections.Generic; using System.Security.Authentication; +using System.Threading.Tasks; using Fleck.Helpers; namespace Fleck @@ -95,7 +96,7 @@ public void Start(Action config) private void ListenForClients() { - ListenerSocket.Accept(OnClientConnect, e => { + var task = ListenerSocket.Accept(OnClientConnect, e => { FleckLog.Error("Listener socket is closed", e); if(RestartAfterListenError){ FleckLog.Info("Listener socket restarting"); @@ -113,14 +114,17 @@ private void ListenForClients() } } }); + + task.ContinueWith (t => FleckLog.Warn ("Client could not connect", t.Exception), TaskContinuationOptions.OnlyOnFaulted); } private void OnClientConnect(ISocket clientSocket) { if (clientSocket == null) return; // socket closed + ListenForClients (); + FleckLog.Debug(String.Format("Client connected from {0}:{1}", clientSocket.RemoteIpAddress, clientSocket.RemotePort.ToString())); - ListenForClients(); WebSocketConnection connection = null; From 3573c0374cddd892bd4ac4fc0aaab1193a0c51e2 Mon Sep 17 00:00:00 2001 From: wmpadmin Date: Sat, 28 Apr 2018 11:04:54 +0200 Subject: [PATCH 2/4] lister-fix refine --- src/Fleck/SocketWrapper.cs | 6 ++-- src/Fleck/WebSocketServer.cs | 63 ++++++++++++++++++++++-------------- 2 files changed, 43 insertions(+), 26 deletions(-) diff --git a/src/Fleck/SocketWrapper.cs b/src/Fleck/SocketWrapper.cs index bee73650..5d8f8fc0 100644 --- a/src/Fleck/SocketWrapper.cs +++ b/src/Fleck/SocketWrapper.cs @@ -132,11 +132,13 @@ public Task Receive(byte[] buffer, Action callback, Action public Task Accept(Action callback, Action error) { - Func end = r => _tokenSource.Token.IsCancellationRequested ? null : new SocketWrapper(_socket.EndAccept(r)); + Func end = r => new SocketWrapper(_socket.EndAccept(r)); var task = _taskFactory.FromAsync(_socket.BeginAccept, end, null); + if (_tokenSource.IsCancellationRequested) return null; + task.ContinueWith(t => error(t.Exception), TaskContinuationOptions.OnlyOnFaulted); - + return task.ContinueWith (t => callback (t.Result), TaskContinuationOptions.OnlyOnRanToCompletion); } diff --git a/src/Fleck/WebSocketServer.cs b/src/Fleck/WebSocketServer.cs index 75bf48ac..a3b41432 100644 --- a/src/Fleck/WebSocketServer.cs +++ b/src/Fleck/WebSocketServer.cs @@ -4,6 +4,7 @@ using System.Security.Cryptography.X509Certificates; using System.Collections.Generic; using System.Security.Authentication; +using System.Threading; using System.Threading.Tasks; using Fleck.Helpers; @@ -94,36 +95,50 @@ public void Start(Action config) _config = config; } - private void ListenForClients() + private void TryRestart () { - var task = ListenerSocket.Accept(OnClientConnect, e => { - FleckLog.Error("Listener socket is closed", e); - if(RestartAfterListenError){ - FleckLog.Info("Listener socket restarting"); - try - { - ListenerSocket.Dispose(); - var socket = new Socket(_locationIP.AddressFamily, SocketType.Stream, ProtocolType.IP); - ListenerSocket = new SocketWrapper(socket); - Start(_config); - FleckLog.Info("Listener socket restarted"); - } - catch (Exception ex) - { - FleckLog.Error("Listener could not be restarted", ex); - } - } - }); - - task.ContinueWith (t => FleckLog.Warn ("Client could not connect", t.Exception), TaskContinuationOptions.OnlyOnFaulted); + FleckLog.Info ("Listener socket restarting"); + try { + ListenerSocket.Dispose (); + var socket = new Socket (_locationIP.AddressFamily, SocketType.Stream, ProtocolType.IP); + ListenerSocket = new SocketWrapper (socket); + Start (_config); + FleckLog.Info ("Listener socket restarted"); + } catch (Exception ex) { + FleckLog.Error ("Listener socket could not be restarted", ex); + } } - private void OnClientConnect(ISocket clientSocket) + + private void ListenForClients () { - if (clientSocket == null) return; // socket closed + ManualResetEvent acceptDone = new ManualResetEvent (false); + + Task.Run (() => { + + while (true) { + + acceptDone.Reset (); + + var task = ListenerSocket.Accept ( + s => { OnClientConnect (s); acceptDone.Set (); }, + e => { FleckLog.Error ("Error while listening for new clients", e); + if (RestartAfterListenError) TryRestart (); + acceptDone.Set (); } + ); - ListenForClients (); + if (task == null) break; + task.ContinueWith((t) => FleckLog.Warn ("Error during client connect", t.Exception), + TaskContinuationOptions.OnlyOnFaulted); + + acceptDone.WaitOne (); + } + }); + } + + private void OnClientConnect(ISocket clientSocket) + { FleckLog.Debug(String.Format("Client connected from {0}:{1}", clientSocket.RemoteIpAddress, clientSocket.RemotePort.ToString())); WebSocketConnection connection = null; From 914fd9c40e08d5a9704449955acac6b0734cf9d8 Mon Sep 17 00:00:00 2001 From: wmpadmin Date: Sat, 28 Apr 2018 11:22:47 +0200 Subject: [PATCH 3/4] refine --- src/Fleck/WebSocketServer.cs | 36 ++++++++++++++++++------------------ 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/src/Fleck/WebSocketServer.cs b/src/Fleck/WebSocketServer.cs index a3b41432..029c29b0 100644 --- a/src/Fleck/WebSocketServer.cs +++ b/src/Fleck/WebSocketServer.cs @@ -109,32 +109,32 @@ private void TryRestart () } } - private void ListenForClients () { - ManualResetEvent acceptDone = new ManualResetEvent (false); + ManualResetEvent acceptDone = new ManualResetEvent (false); + bool running = true; - Task.Run (() => { - - while (true) { + Task.Run (() => { - acceptDone.Reset (); + while (running) { + + acceptDone.Reset (); - var task = ListenerSocket.Accept ( - s => { OnClientConnect (s); acceptDone.Set (); }, - e => { FleckLog.Error ("Error while listening for new clients", e); - if (RestartAfterListenError) TryRestart (); - acceptDone.Set (); } - ); + var task = ListenerSocket.Accept( + s => { acceptDone.Set (); OnClientConnect (s); }, + e => { FleckLog.Error ("Error while listening for new clients", e); + if (RestartAfterListenError) TryRestart (); + running = false; acceptDone.Set (); } + ); - if (task == null) break; + if(task == null) break; - task.ContinueWith((t) => FleckLog.Warn ("Error during client connect", t.Exception), - TaskContinuationOptions.OnlyOnFaulted); + task.ContinueWith((t) => FleckLog.Warn ("Error during client connect", t.Exception), + TaskContinuationOptions.OnlyOnFaulted); - acceptDone.WaitOne (); - } - }); + acceptDone.WaitOne (); + } + }); } private void OnClientConnect(ISocket clientSocket) From 06eebf1856d73869473b0993b90631bce293eb65 Mon Sep 17 00:00:00 2001 From: wmpadmin Date: Sat, 28 Apr 2018 11:52:03 +0200 Subject: [PATCH 4/4] testcase fix --- src/Fleck/SocketWrapper.cs | 4 +-- src/Fleck/WebSocketServer.cs | 47 +++++++++++++++++++----------------- 2 files changed, 26 insertions(+), 25 deletions(-) diff --git a/src/Fleck/SocketWrapper.cs b/src/Fleck/SocketWrapper.cs index 5d8f8fc0..91d35bc9 100644 --- a/src/Fleck/SocketWrapper.cs +++ b/src/Fleck/SocketWrapper.cs @@ -132,10 +132,8 @@ public Task Receive(byte[] buffer, Action callback, Action public Task Accept(Action callback, Action error) { - Func end = r => new SocketWrapper(_socket.EndAccept(r)); + Func end = r => _tokenSource.Token.IsCancellationRequested ? null : new SocketWrapper (_socket.EndAccept (r)); var task = _taskFactory.FromAsync(_socket.BeginAccept, end, null); - - if (_tokenSource.IsCancellationRequested) return null; task.ContinueWith(t => error(t.Exception), TaskContinuationOptions.OnlyOnFaulted); diff --git a/src/Fleck/WebSocketServer.cs b/src/Fleck/WebSocketServer.cs index 029c29b0..9594e266 100644 --- a/src/Fleck/WebSocketServer.cs +++ b/src/Fleck/WebSocketServer.cs @@ -111,34 +111,37 @@ private void TryRestart () private void ListenForClients () { - ManualResetEvent acceptDone = new ManualResetEvent (false); - bool running = true; + ManualResetEvent acceptDone = new ManualResetEvent (false); + bool running = true; - Task.Run (() => { + Task.Run (() => { + + while (running) { - while (running) { - - acceptDone.Reset (); - - var task = ListenerSocket.Accept( - s => { acceptDone.Set (); OnClientConnect (s); }, - e => { FleckLog.Error ("Error while listening for new clients", e); - if (RestartAfterListenError) TryRestart (); - running = false; acceptDone.Set (); } - ); - - if(task == null) break; - - task.ContinueWith((t) => FleckLog.Warn ("Error during client connect", t.Exception), - TaskContinuationOptions.OnlyOnFaulted); - - acceptDone.WaitOne (); - } - }); + acceptDone.Reset (); + + var task = ListenerSocket.Accept( + s => { + running = (s != null); + acceptDone.Set (); + OnClientConnect (s); }, + e => { FleckLog.Error ("Error while listening for new clients", e); + if (RestartAfterListenError) TryRestart (); + running = false; acceptDone.Set (); } + ); + + task.ContinueWith((t) => FleckLog.Warn ("Error during client connect", t.Exception), + TaskContinuationOptions.OnlyOnFaulted); + + acceptDone.WaitOne (); + } + }); } private void OnClientConnect(ISocket clientSocket) { + if (clientSocket == null) return; // socket closed + FleckLog.Debug(String.Format("Client connected from {0}:{1}", clientSocket.RemoteIpAddress, clientSocket.RemotePort.ToString())); WebSocketConnection connection = null;