diff --git a/MTSC.UnitTests/E2ETests.cs b/MTSC.UnitTests/E2ETests.cs index 0ba657c..5bc925a 100644 --- a/MTSC.UnitTests/E2ETests.cs +++ b/MTSC.UnitTests/E2ETests.cs @@ -41,20 +41,38 @@ namespace MTSC.UnitTests websocketMessage.Opcode = WebsocketMessage.Opcodes.Text; return websocketMessage; }))) - .AddHandler(new HttpHandler() - .AddHttpModule(new HttpRoutingModule() - .AddRoute(HttpMessage.HttpMethods.Get, "", new Http200Module()) - .AddRoute(HttpMessage.HttpMethods.Get, "query", new TestQueryModule()) - .AddRoute(HttpMessage.HttpMethods.Get, "echo", new EchoModule())) + .AddHandler(new HttpRoutingHandler() + .AddRoute(HttpMessage.HttpMethods.Get, "", new Http200Module()) + .AddRoute(HttpMessage.HttpMethods.Get, "query", new TestQueryModule()) + .AddRoute(HttpMessage.HttpMethods.Get, "echo", new EchoModule()) + .AddRoute(HttpMessage.HttpMethods.Get, "long-running", new LongRunningModule()) .WithFragmentsExpirationTime(TimeSpan.FromSeconds(1)) .WithMaximumSize(300)) .AddLogger(new ConsoleLogger()) .AddLogger(new DebugConsoleLogger()) .AddExceptionHandler(new ExceptionConsoleLogger()) - .SetScheduler(new SequentialProcessingScheduler()) + .SetScheduler(new FireTasksAndForgetScheduler()) .WithSslAuthenticationTimeout(TimeSpan.FromMilliseconds(100)); Server.RunAsync(); } + [TestMethod] + public async Task ServerRespondsDuringLongRunningTask() + { + HttpClient httpClient = new HttpClient(); + httpClient.BaseAddress = new Uri("http://localhost:800"); + var longRunningTask = httpClient.GetAsync("long-running"); + int responses = 0; + HttpClient client2 = new HttpClient(); + client2.BaseAddress = new Uri("http://localhost:800"); + while (!longRunningTask.IsCompleted) + { + var echoResponse = await client2.GetAsync("echo"); + responses++; + } + var result = longRunningTask.Result; + Assert.AreEqual(result.StatusCode, System.Net.HttpStatusCode.OK); + Assert.IsTrue(responses > 50); + } [TestMethod] public void HelloWorldHTTP() diff --git a/MTSC.UnitTests/EchoModule.cs b/MTSC.UnitTests/EchoModule.cs index 26a48f3..801dad0 100644 --- a/MTSC.UnitTests/EchoModule.cs +++ b/MTSC.UnitTests/EchoModule.cs @@ -1,14 +1,15 @@ using MTSC.Common.Http; using MTSC.Common.Http.RoutingModules; using MTSC.ServerSide; +using System.Threading.Tasks; namespace MTSC.UnitTests { public class EchoModule : HttpRouteBase { - public override HttpResponse HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) + public override Task HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) { - return new HttpResponse { BodyString = request.BodyString, StatusCode = HttpMessage.StatusCodes.OK }; + return Task.FromResult(new HttpResponse { BodyString = request.BodyString, StatusCode = HttpMessage.StatusCodes.OK }); } } } diff --git a/MTSC.UnitTests/LongRunningModule.cs b/MTSC.UnitTests/LongRunningModule.cs new file mode 100644 index 0000000..f5acdbd --- /dev/null +++ b/MTSC.UnitTests/LongRunningModule.cs @@ -0,0 +1,16 @@ +using MTSC.Common.Http; +using MTSC.Common.Http.RoutingModules; +using MTSC.ServerSide; +using System.Threading.Tasks; + +namespace MTSC.UnitTests +{ + public class LongRunningModule : HttpRouteBase + { + public async override Task HandleRequest(HttpRequest request, ClientData client, Server server) + { + await Task.Delay(5000); + return new HttpResponse { StatusCode = HttpMessage.StatusCodes.OK }; + } + } +} diff --git a/MTSC.UnitTests/TestQueryModule.cs b/MTSC.UnitTests/TestQueryModule.cs index be2f8aa..5c6e18e 100644 --- a/MTSC.UnitTests/TestQueryModule.cs +++ b/MTSC.UnitTests/TestQueryModule.cs @@ -1,24 +1,25 @@ using MTSC.Common.Http; using MTSC.Common.Http.RoutingModules; using MTSC.ServerSide; +using System.Threading.Tasks; using System.Web; namespace MTSC.UnitTests { class TestQueryModule : HttpRouteBase { - public override HttpResponse HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) + public override Task HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) { var query = HttpUtility.ParseQueryString(request.RequestQuery); if(query.Count == 2 && query.Keys[0] == "key1" && query.Keys[1] == "key2") { - return new HttpResponse { StatusCode = HttpMessage.StatusCodes.OK }; + return Task.FromResult(new HttpResponse { StatusCode = HttpMessage.StatusCodes.OK }); } else { - return new HttpResponse { StatusCode = HttpMessage.StatusCodes.BadRequest }; + return Task.FromResult(new HttpResponse { StatusCode = HttpMessage.StatusCodes.BadRequest }); } } } diff --git a/MTSC/Common/Http/RoutingModules/Http200Module.cs b/MTSC/Common/Http/RoutingModules/Http200Module.cs index 2d80e66..2af4fc7 100644 --- a/MTSC/Common/Http/RoutingModules/Http200Module.cs +++ b/MTSC/Common/Http/RoutingModules/Http200Module.cs @@ -1,12 +1,13 @@ using MTSC.ServerSide; +using System.Threading.Tasks; namespace MTSC.Common.Http.RoutingModules { public sealed class Http200Module : HttpRouteBase { - public override HttpResponse HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) + public override Task HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) { - return new HttpResponse { StatusCode = HttpMessage.StatusCodes.OK }; + return Task.FromResult(new HttpResponse { StatusCode = HttpMessage.StatusCodes.OK }); } } } diff --git a/MTSC/Common/Http/RoutingModules/HttpRouteBase.cs b/MTSC/Common/Http/RoutingModules/HttpRouteBase.cs index 7de0941..f0e9ebb 100644 --- a/MTSC/Common/Http/RoutingModules/HttpRouteBase.cs +++ b/MTSC/Common/Http/RoutingModules/HttpRouteBase.cs @@ -1,16 +1,17 @@ using MTSC.ServerSide; using System; +using System.Threading.Tasks; namespace MTSC.Common.Http.RoutingModules { public abstract class HttpRouteBase { - public HttpResponse CallHandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) + public async Task CallHandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) { - return this.HandleRequest(request, client, server); + return await this.HandleRequest(request, client, server); } - public abstract HttpResponse HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server); + public abstract Task HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server); } public abstract class HttpRouteBase : HttpRouteBase { @@ -32,12 +33,12 @@ namespace MTSC.Common.Http.RoutingModules return this; } - public override HttpResponse HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) + public async override Task HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) { - return HandleRequest(template.Invoke(request), client, server); + return await HandleRequest(template.Invoke(request), client, server); } - public abstract HttpResponse HandleRequest(T request, ClientData client, ServerSide.Server server); + public abstract Task HandleRequest(T request, ClientData client, ServerSide.Server server); } public abstract class HttpRouteBase : HttpRouteBase { @@ -67,11 +68,11 @@ namespace MTSC.Common.Http.RoutingModules return this; } - public override HttpResponse HandleRequest(HttpRequest request, ClientData client, ServerSide.Server server) + public async override Task HandleRequest(HttpRequest request, ClientData client, Server server) { - return sendTemplate.Invoke(HandleRequest(receiveTemplate.Invoke(request), client, server)); + return sendTemplate.Invoke(await HandleRequest(receiveTemplate.Invoke(request), client, server)); } - public abstract TSend HandleRequest(TReceive request, ClientData client, ServerSide.Server server); + public abstract Task HandleRequest(TReceive request, ClientData client, Server server); } } diff --git a/MTSC/Common/Http/ServerModules/HttpRoutingModule.cs b/MTSC/Common/Http/ServerModules/HttpRoutingModule.cs deleted file mode 100644 index c334d0e..0000000 --- a/MTSC/Common/Http/ServerModules/HttpRoutingModule.cs +++ /dev/null @@ -1,94 +0,0 @@ -using MTSC.Common.Http.RoutingModules; -using MTSC.ServerSide; -using MTSC.ServerSide.Handlers; -using System; -using System.Collections.Generic; -using static MTSC.Common.Http.HttpMessage; - -namespace MTSC.Common.Http.ServerModules -{ - public sealed class HttpRoutingModule : IHttpModule - { - private static Func alwaysEnabled = (server, request, client) => RouteEnablerResponse.Accept; - - private Dictionary)>> moduleDictionary = - new Dictionary)>>(); - - public HttpRoutingModule() - { - foreach (HttpMethods method in (HttpMethods[])Enum.GetValues(typeof(HttpMethods))) - { - moduleDictionary[method] = new Dictionary)>(); - } - } - - public HttpRoutingModule AddRoute( - HttpMethods method, - string uri, - HttpRouteBase routeModule) - { - moduleDictionary[method][uri] = (routeModule, alwaysEnabled); - return this; - } - - public HttpRoutingModule AddRoute( - HttpMethods method, - string uri, - HttpRouteBase routeModule, - Func routeEnabler) - { - moduleDictionary[method][uri] = (routeModule, routeEnabler); - return this; - } - - public HttpRoutingModule RemoveRoute( - HttpMethods method, - string uri) - { - moduleDictionary[method].Remove(uri); - return this; - } - - bool IHttpModule.HandleRequest(ServerSide.Server server, HttpHandler handler, ClientData client, HttpRequest request, ref HttpResponse response) - { - /* - * Now find if a routing module exists. If not let other handlers try and handle the message. - */ - if (moduleDictionary[request.Method].ContainsKey(request.RequestURI)) - { - (var module, var routeEnabler) = moduleDictionary[request.Method][request.RequestURI]; - var routeEnablerResponse = routeEnabler.Invoke(server, request, client); - if (routeEnablerResponse is RouteEnablerResponse.RouteEnablerResponseAccept) - { - try - { - response = module.CallHandleRequest(request, client, server); - //response = module.HandleRequest(requestTemplate.Invoke(request), client, server); - } - catch(Exception e) - { - server.LogDebug("Exception: " + e.Message); - server.LogDebug("Stacktrace: " + e.StackTrace); - response = new HttpResponse() { StatusCode = StatusCodes.InternalServerError }; - } - return true; - } - else if(routeEnablerResponse is RouteEnablerResponse.RouteEnablerResponseIgnore) - { - return false; - } - else if(routeEnablerResponse is RouteEnablerResponse.RouteEnablerResponseError) - { - response = (routeEnablerResponse as RouteEnablerResponse.RouteEnablerResponseError).Response; - return true; - } - } - return false; - } - - void IHttpModule.Tick(ServerSide.Server server, HttpHandler handler) { } - } -} diff --git a/MTSC/Common/Http/ServerModules/IHttpModule.cs b/MTSC/Common/Http/ServerModules/IHttpModule.cs index bade6d2..5e50a86 100644 --- a/MTSC/Common/Http/ServerModules/IHttpModule.cs +++ b/MTSC/Common/Http/ServerModules/IHttpModule.cs @@ -15,11 +15,11 @@ namespace MTSC.Common.Http.ServerModules /// Client data. /// Request message. /// True if no other module should handle the received request. - bool HandleRequest(ServerSide.Server server, HttpHandler handler, ClientData client, HttpRequest request, ref HttpResponse response); + bool HandleRequest(Server server, HttpHandler handler, ClientData client, HttpRequest request, ref HttpResponse response); /// /// Perform periodic operations. /// /// - void Tick(ServerSide.Server server, HttpHandler handler); + void Tick(Server server, HttpHandler handler); } } diff --git a/MTSC/Common/IConsumerQueue.cs b/MTSC/Common/IConsumerQueue.cs new file mode 100644 index 0000000..aadabf1 --- /dev/null +++ b/MTSC/Common/IConsumerQueue.cs @@ -0,0 +1,18 @@ +namespace MTSC.Common +{ + public interface IConsumerQueue + { + /// + /// Dequeues the first element from the queue + /// + /// + T Dequeue(); + /// + /// Tries to dequeue the first element from the queue. is set to the proper value + /// if the operation succeeds. Otherwise, it is set to default value of + /// + /// + /// True if the operation succeeds. False if the operations has failed + bool TryDequeue(out T value); + } +} diff --git a/MTSC/Common/IProducerQueue.cs b/MTSC/Common/IProducerQueue.cs new file mode 100644 index 0000000..26fed02 --- /dev/null +++ b/MTSC/Common/IProducerQueue.cs @@ -0,0 +1,12 @@ +namespace MTSC.Common +{ + public interface IProducerQueue + { + /// + /// Enqueues the value to the queue + /// + /// + /// + void Enqueue(T value); + } +} diff --git a/MTSC/Common/ProducerConsumerQueue.cs b/MTSC/Common/ProducerConsumerQueue.cs new file mode 100644 index 0000000..56b92b4 --- /dev/null +++ b/MTSC/Common/ProducerConsumerQueue.cs @@ -0,0 +1,44 @@ +using System; +using System.Collections.Generic; + +namespace MTSC.Common +{ + public class ProducerConsumerQueue : IProducerQueue, IConsumerQueue + { + private Queue queue = new Queue(); + + TValue IConsumerQueue.Dequeue() + { + lock (queue) + { + if (queue.Count > 0) return queue.Dequeue(); + else throw new InvalidOperationException("There are no elements to dequeue from the queue"); + } + } + + bool IConsumerQueue.TryDequeue(out TValue value) + { + lock (queue) + { + if (queue.Count > 0) + { + value = queue.Dequeue(); + return true; + } + else + { + value = default; + return false; + } + } + } + + void IProducerQueue.Enqueue(TValue value) + { + lock (queue) + { + queue.Enqueue(value); + } + } + } +} diff --git a/MTSC/MTSC.csproj b/MTSC/MTSC.csproj index 6d6eaea..c47d524 100644 --- a/MTSC/MTSC.csproj +++ b/MTSC/MTSC.csproj @@ -5,12 +5,12 @@ netcoreapp2.1;net48;netstandard2.0;netcoreapp3.0 - 2.2.5 + 2.3 Alexandru-Victor Macocian MTSC Modular TCP Server and Client - 0.2.2.5 - 0.2.2.5 + 0.2.3.0 + 0.2.3.0 true AnyCPU;x64 https://github.com/AlexMacocian/MTSC diff --git a/MTSC/ServerSide/Handlers/HttpRoutingHandler.cs b/MTSC/ServerSide/Handlers/HttpRoutingHandler.cs new file mode 100644 index 0000000..fecdf9f --- /dev/null +++ b/MTSC/ServerSide/Handlers/HttpRoutingHandler.cs @@ -0,0 +1,234 @@ +using MTSC.Common.Http; +using MTSC.Common.Http.RoutingModules; +using MTSC.Exceptions; +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using static MTSC.Common.Http.HttpMessage; + +namespace MTSC.ServerSide.Handlers +{ + public sealed class HttpRoutingHandler : IHandler + { + private static Func alwaysEnabled = (server, request, client) => RouteEnablerResponse.Accept; + + private List removeFragmentsList = new List(); + private ConcurrentQueue> messageOutQueue = new ConcurrentQueue>(); + private ConcurrentDictionary fragmentedMessages = new ConcurrentDictionary(); + + private Dictionary)>> moduleDictionary = + new Dictionary)>>(); + + public TimeSpan FragmentsExpirationTime { get; set; } = TimeSpan.FromSeconds(15); + public double MaximumRequestSize { get; set; } = 15000; + + public HttpRoutingHandler() + { + foreach (HttpMethods method in (HttpMethods[])Enum.GetValues(typeof(HttpMethods))) + { + moduleDictionary[method] = new Dictionary)>(); + } + } + + public HttpRoutingHandler AddRoute( + HttpMethods method, + string uri, + HttpRouteBase routeModule) + { + moduleDictionary[method][uri] = (routeModule, alwaysEnabled); + return this; + } + public HttpRoutingHandler AddRoute( + HttpMethods method, + string uri, + HttpRouteBase routeModule, + Func routeEnabler) + { + moduleDictionary[method][uri] = (routeModule, routeEnabler); + return this; + } + public HttpRoutingHandler RemoveRoute( + HttpMethods method, + string uri) + { + moduleDictionary[method].Remove(uri); + return this; + } + + public HttpRoutingHandler WithMaximumSize(double size) + { + this.MaximumRequestSize = size; + return this; + } + /// + /// The amount of time fragments are kept in the buffer before being discarded. + /// + /// Time until fragments expire. + /// This handler object. + public HttpRoutingHandler WithFragmentsExpirationTime(TimeSpan duration) + { + this.FragmentsExpirationTime = duration; + return this; + } + /// + /// Send a response back to the client. + /// + /// Message containing the response. + public void QueueResponse(ClientData client, HttpResponse response) + { + messageOutQueue.Enqueue(new Tuple(client, response)); + } + + void IHandler.ClientRemoved(Server server, ClientData client) { } + + bool IHandler.HandleClient(Server server, ClientData client) => false; + + bool IHandler.HandleReceivedMessage(Server server, ClientData client, Message message) + { + // Parse the request. If the message is incomplete, return 100 and queue the message to be parsed later. + HttpRequest request = null; + byte[] messageBytes = null; + try + { + if (fragmentedMessages.ContainsKey(client)) + { + byte[] previousBytes = fragmentedMessages[client].Item1; + if (previousBytes.Length + message.MessageBytes.Length > MaximumRequestSize) + { + // Discard the message if it is too big + server.LogDebug($"Discarded message. Message size [{previousBytes.Length + message.MessageBytes.Length}] > [{MaximumRequestSize}]"); + fragmentedMessages.TryRemove(client, out _); + return false; + } + byte[] repackagingBuffer = new byte[previousBytes.Length + message.MessageBytes.Length]; + Array.Copy(previousBytes, 0, repackagingBuffer, 0, previousBytes.Length); + Array.Copy(message.MessageBytes, 0, repackagingBuffer, previousBytes.Length, message.MessageBytes.Length); + messageBytes = repackagingBuffer; + } + else + { + if (message.MessageBytes.Length > MaximumRequestSize) + { + // Discard the message if it is too big + server.LogDebug($"Discarded message. Message size [{message.MessageBytes.Length}] > [{MaximumRequestSize}]"); + return false; + } + messageBytes = message.MessageBytes; + } + messageBytes = messageBytes.TrimTrailingNullBytes(); + var partialRequest = PartialHttpRequest.FromBytes(messageBytes); + if (partialRequest.Complete) + request = partialRequest.ToRequest(); + else + { + HandleIncompleteRequest(client, server, messageBytes, partialRequest); + return true; + } + } + catch (Exception ex) when ( + ex is IncompleteHeaderKeyException || + ex is IncompleteHeaderValueException || + ex is IncompleteHttpVersionException || + ex is IncompleteMethodException || + ex is IncompleteRequestBodyException || + ex is IncompleteRequestQueryException || + ex is IncompleteRequestURIException || + ex is IncompleteRequestException || + ex is InvalidPostFormException) + { + server.LogDebug(ex.Message); + server.LogDebug(ex.StackTrace); + HandleIncompleteRequest(client, server, messageBytes); + return true; + } + catch (Exception e) + { + throw e; + } + + // The message has been parsed. If there was a cache for the current message, remove it. + if (fragmentedMessages.ContainsKey(client)) + { + fragmentedMessages.TryRemove(client, out _); + } + + /* + * Now find if a routing module exists. If not let other handlers try and handle the message. + */ + if (moduleDictionary[request.Method].ContainsKey(request.RequestURI)) + { + (var module, var routeEnabler) = moduleDictionary[request.Method][request.RequestURI]; + var routeEnablerResponse = routeEnabler.Invoke(server, request, client); + if (routeEnablerResponse is RouteEnablerResponse.RouteEnablerResponseAccept) + { + try + { + module.CallHandleRequest(request, client, server).ContinueWith((task) => { QueueResponse(client, task.Result); }); + //response = module.HandleRequest(requestTemplate.Invoke(request), client, server); + } + catch (Exception e) + { + server.LogDebug("Exception: " + e.Message); + server.LogDebug("Stacktrace: " + e.StackTrace); + QueueResponse(client, new HttpResponse() { StatusCode = StatusCodes.InternalServerError }); + } + return true; + } + else if (routeEnablerResponse is RouteEnablerResponse.RouteEnablerResponseIgnore) + { + return false; + } + else if (routeEnablerResponse is RouteEnablerResponse.RouteEnablerResponseError) + { + QueueResponse(client, (routeEnablerResponse as RouteEnablerResponse.RouteEnablerResponseError).Response); + return true; + } + } + return false; + } + + bool IHandler.HandleSendMessage(Server server, ClientData client, ref Message message) => false; + + bool IHandler.PreHandleReceivedMessage(Server server, ClientData client, ref Message message) => false; + + void IHandler.Tick(Server server) + { + while (messageOutQueue.Count > 0) + { + if (messageOutQueue.TryDequeue(out Tuple tuple)) + { + server.QueueMessage(tuple.Item1, tuple.Item2.GetPackedResponse(true)); + } + } + removeFragmentsList.Clear(); + foreach (var kvp in fragmentedMessages) + { + if ((DateTime.Now - kvp.Value.Item2) > FragmentsExpirationTime) + { + removeFragmentsList.Add(kvp.Key); + } + } + foreach (var key in removeFragmentsList) + { + fragmentedMessages.TryRemove(key, out _); + } + } + + private void HandleIncompleteRequest(ClientData client, Server server, byte[] messageBytes, PartialHttpRequest partialRequest = null) + { + fragmentedMessages[client] = (messageBytes, DateTime.Now); + server.LogDebug("Incomplete request received!"); + if (partialRequest != null && partialRequest.Headers.ContainsHeader(HttpMessage.RequestHeaders.Expect) && + partialRequest.Headers[HttpMessage.RequestHeaders.Expect].Equals("100-continue", StringComparison.OrdinalIgnoreCase)) + { + server.LogDebug("Returning 100-Continue"); + var contResponse = new HttpResponse { StatusCode = HttpMessage.StatusCodes.Continue }; + contResponse.Headers[HttpMessage.GeneralHeaders.Connection] = "keep-alive"; + QueueResponse(client, contResponse); + } + } + } +} diff --git a/MTSC/ServerSide/Schedulers/FireTasksAndForgetScheduler.cs b/MTSC/ServerSide/Schedulers/FireTasksAndForgetScheduler.cs index 7ba08b4..dd80ab0 100644 --- a/MTSC/ServerSide/Schedulers/FireTasksAndForgetScheduler.cs +++ b/MTSC/ServerSide/Schedulers/FireTasksAndForgetScheduler.cs @@ -1,14 +1,14 @@ -using System; -using System.Collections.Concurrent; +using MTSC.Common; +using System; using System.Threading.Tasks; namespace MTSC.ServerSide.Schedulers { public class FireTasksAndForgetScheduler : IScheduler { - public void ScheduleHandling(IProducerConsumerCollection<(ClientData, Message)> inQueue, Action messageHandlingProcedure) + public void ScheduleHandling(IConsumerQueue<(ClientData, Message)> inQueue, Action messageHandlingProcedure) { - while(inQueue.TryTake(out var tuple)) + while(inQueue.TryDequeue(out var tuple)) { (var client, var message) = tuple; Task.Run(() => { messageHandlingProcedure.Invoke(client, message); }); diff --git a/MTSC/ServerSide/Schedulers/IScheduler.cs b/MTSC/ServerSide/Schedulers/IScheduler.cs index 767070a..03d24d3 100644 --- a/MTSC/ServerSide/Schedulers/IScheduler.cs +++ b/MTSC/ServerSide/Schedulers/IScheduler.cs @@ -1,10 +1,10 @@ -using System; -using System.Collections.Concurrent; +using MTSC.Common; +using System; namespace MTSC.ServerSide.Schedulers { public interface IScheduler { - void ScheduleHandling(IProducerConsumerCollection<(ClientData, Message)> inQueue, Action messageHandlingProcedure); + void ScheduleHandling(IConsumerQueue<(ClientData, Message)> inQueue, Action messageHandlingProcedure); } } diff --git a/MTSC/ServerSide/Schedulers/ParallelScheduler.cs b/MTSC/ServerSide/Schedulers/ParallelScheduler.cs index dfe94b1..c14fbab 100644 --- a/MTSC/ServerSide/Schedulers/ParallelScheduler.cs +++ b/MTSC/ServerSide/Schedulers/ParallelScheduler.cs @@ -1,5 +1,5 @@ -using System; -using System.Collections.Concurrent; +using MTSC.Common; +using System; using System.Collections.Generic; using System.Threading.Tasks; @@ -7,11 +7,11 @@ namespace MTSC.ServerSide.Schedulers { public class ParallelScheduler : IScheduler { - void IScheduler.ScheduleHandling(IProducerConsumerCollection<(ClientData, Message)> inQueue, Action messageHandlingProcedure) + void IScheduler.ScheduleHandling(IConsumerQueue<(ClientData, Message)> inQueue, Action messageHandlingProcedure) { - List actionList = new List(inQueue.Count); + List actionList = new List(); - while(inQueue.TryTake(out var tuple)) + while(inQueue.TryDequeue(out var tuple)) { (var client, var message) = tuple; actionList.Add(new Action(() => { messageHandlingProcedure.Invoke(client, message); })); diff --git a/MTSC/ServerSide/Schedulers/SequentialProcessingScheduler.cs b/MTSC/ServerSide/Schedulers/SequentialProcessingScheduler.cs index edb0ffa..90f2732 100644 --- a/MTSC/ServerSide/Schedulers/SequentialProcessingScheduler.cs +++ b/MTSC/ServerSide/Schedulers/SequentialProcessingScheduler.cs @@ -1,13 +1,14 @@ -using System; +using MTSC.Common; +using System; using System.Collections.Concurrent; namespace MTSC.ServerSide.Schedulers { public class SequentialProcessingScheduler : IScheduler { - void IScheduler.ScheduleHandling(IProducerConsumerCollection<(ClientData, Message)> inQueue, Action messageHandlingProcedure) + void IScheduler.ScheduleHandling(IConsumerQueue<(ClientData, Message)> inQueue, Action messageHandlingProcedure) { - while (inQueue.TryTake(out var tuple)) + while (inQueue.TryDequeue(out var tuple)) { (var client, var message) = tuple; messageHandlingProcedure.Invoke(client, message); diff --git a/MTSC/ServerSide/Schedulers/TaskAwaiterScheduler.cs b/MTSC/ServerSide/Schedulers/TaskAwaiterScheduler.cs index 4debaaf..e6d3d0a 100644 --- a/MTSC/ServerSide/Schedulers/TaskAwaiterScheduler.cs +++ b/MTSC/ServerSide/Schedulers/TaskAwaiterScheduler.cs @@ -1,5 +1,5 @@ -using System; -using System.Collections.Concurrent; +using MTSC.Common; +using System; using System.Collections.Generic; using System.Threading.Tasks; @@ -7,10 +7,10 @@ namespace MTSC.ServerSide.Schedulers { public class TaskAwaiterScheduler : IScheduler { - void IScheduler.ScheduleHandling(IProducerConsumerCollection<(ClientData, Message)> inQueue, Action messageHandlingProcedure) + void IScheduler.ScheduleHandling(IConsumerQueue<(ClientData, Message)> inQueue, Action messageHandlingProcedure) { List tasks = new List(); - while(inQueue.TryTake(out var tuple)) + while(inQueue.TryDequeue(out var tuple)) { (var client, var message) = tuple; tasks.Add(Task.Run(() => messageHandlingProcedure.Invoke(client, message))); diff --git a/MTSC/ServerSide/Server.cs b/MTSC/ServerSide/Server.cs index 47234b4..cb9a801 100644 --- a/MTSC/ServerSide/Server.cs +++ b/MTSC/ServerSide/Server.cs @@ -1,4 +1,5 @@ -using MTSC.Exceptions; +using MTSC.Common; +using MTSC.Exceptions; using MTSC.Logging; using MTSC.ServerSide.Handlers; using MTSC.ServerSide.Resources; @@ -25,16 +26,36 @@ namespace MTSC.ServerSide bool running; X509Certificate2 certificate; TcpListener listener; + ProducerConsumerQueue addQueue = new ProducerConsumerQueue(); + List clients = new List(); List toRemove = new List(); List handlers = new List(); List loggers = new List(); List exceptionHandlers = new List(); List serverUsageMonitors = new List(); + ProducerConsumerQueue<(ClientData, Message)> messageInQueue = new ProducerConsumerQueue<(ClientData, Message)>(); + ProducerConsumerQueue<(ClientData, byte[])> messageOutQueue = new ProducerConsumerQueue<(ClientData, byte[])>(); #endregion - #region Properties + #region Private Properties + private IConsumerQueue _ConsumerClientQueue { get => addQueue; } + private IProducerQueue _ProducerClientQueue { get => addQueue; } + private IConsumerQueue<(ClientData, Message)> _ConsumerMessageInQueue { get => messageInQueue; } + private IConsumerQueue<(ClientData, byte[])> _ConsumerMessageOutQueue { get => messageOutQueue; } + private IProducerQueue<(ClientData, Message)> _ProducerMessageInQueue { get => messageInQueue; } + #endregion + #region Public Properties + /// + /// Client handling scheduler + /// public IScheduler Scheduler { get; set; } = new ParallelScheduler(); - public IProducerConsumerCollection<(ClientData, Message)> InQueue { get; set; } = new ConcurrentQueue<(ClientData, Message)>(); - public IProducerConsumerCollection<(ClientData, byte[])> OutQueue { get; set; } = new ConcurrentQueue<(ClientData, byte[])>(); + + /// + /// Queue of destinations and messages to be processed + /// + public IProducerQueue<(ClientData, byte[])> MessageOutQueue { get => messageOutQueue; } + /// + /// Duration until ssl authentication gives up during the authentication process + /// public TimeSpan SslAuthenticationTimeout { get; set; } = TimeSpan.FromSeconds(1); /// /// SSL supported protocols. @@ -64,7 +85,7 @@ namespace MTSC.ServerSide /// /// List of clients currently connected to the server. /// - public List Clients { get; set; } = new List(); + public IReadOnlyCollection Clients { get => clients.AsReadOnly(); } /// /// Dictionary of resources /// @@ -109,32 +130,6 @@ namespace MTSC.ServerSide return this; } /// - /// Sets the InQueue with the provided type. - /// Default type is a . - /// Modifying the queue will have an impact on performance, depending on the implementation of the provided collection. - /// - /// - /// - /// This server object. - public Server SetInQueue(T queue) where T : IProducerConsumerCollection<(ClientData, Message)> - { - this.InQueue = queue; - return this; - } - /// - /// Sets the OutQueue with the provided type. - /// Default type is a . - /// Modifying the queue will have an impact on performance, depending on the implementation of the provided collection. - /// - /// - /// - /// This server object. - public Server SetOutQueue(T queue) where T : IProducerConsumerCollection<(ClientData, byte[])> - { - this.OutQueue = queue; - return this; - } - /// /// Ssl authentication timeout /// /// @@ -339,15 +334,7 @@ namespace MTSC.ServerSide /// Message to be sent. public void QueueMessage(ClientData target, byte[] message) { - int retries = 0; - while(!OutQueue.TryAdd((target, message))) - { - retries++; - if(retries > 5) - { - throw new QueueOperationException($"Failed to insert provided message in the {nameof(OutQueue)}. Tried {retries} times"); - } - }; + (MessageOutQueue as IProducerQueue<(ClientData, byte[])>).Enqueue((target, message)); } /// /// Adds a message to be logged by the associated loggers. @@ -419,7 +406,7 @@ namespace MTSC.ServerSide { TcpClient tcpClient = listener.AcceptTcpClient(); ClientData clientStruct = new ClientData(tcpClient); - AcceptClient(clientStruct); + Task.Run(() => AcceptClient(clientStruct)); } } catch (Exception e) @@ -432,6 +419,22 @@ namespace MTSC.ServerSide } } } + /* + * Add all accepted clients to the list + */ + while(this._ConsumerClientQueue.TryDequeue(out var client)) + { + Log("Accepted new connection: " + client.TcpClient.Client.RemoteEndPoint.ToString()); + clients.Add(client); + foreach (IHandler handler in handlers) + { + if (handler.HandleClient(this, client)) + { + break; + } + } + } + /* * Gather all messages from clients and put them in a queue */ @@ -441,7 +444,7 @@ namespace MTSC.ServerSide * Call the scheduler to handle all received messages and distribute them to the handlers */ - Scheduler.ScheduleHandling(InQueue, HandleClientMessage); + Scheduler.ScheduleHandling(this._ConsumerMessageInQueue, HandleClientMessage); /* * Iterate through all the handlers, running periodic operations. @@ -516,25 +519,22 @@ namespace MTSC.ServerSide #region Private Methods private void SendQueuedMessages() { - while (OutQueue.Count > 0) + while (this._ConsumerMessageOutQueue.TryDequeue(out var tuple)) { - try + (var client, var bytes) = tuple; + try { - if (OutQueue.TryTake(out (ClientData, byte[]) queuedOrder)) + Message sendMessage = CommunicationPrimitives.BuildMessage(bytes); + for (int i = handlers.Count - 1; i >= 0; i--) { - Message sendMessage = CommunicationPrimitives.BuildMessage(queuedOrder.Item2); - for (int i = handlers.Count - 1; i >= 0; i--) - { - IHandler handler = handlers[i]; - ClientData client = queuedOrder.Item1; - handler.HandleSendMessage(this, client, ref sendMessage); - } - CommunicationPrimitives.SendMessage(queuedOrder.Item1.TcpClient, sendMessage, queuedOrder.Item1.SslStream); - LogDebug("Sent message to " + queuedOrder.Item1.TcpClient.Client.RemoteEndPoint.ToString() + - "\nMessage length: " + sendMessage.MessageLength); + IHandler handler = handlers[i]; + handler.HandleSendMessage(this, client, ref sendMessage); } + CommunicationPrimitives.SendMessage(client.TcpClient, sendMessage, client.SslStream); + LogDebug("Sent message to " + client.TcpClient.Client.RemoteEndPoint.ToString() + + "\nMessage length: " + sendMessage.MessageLength); } - catch (Exception e) + catch(Exception e) { foreach (IExceptionHandler exceptionHandler in exceptionHandlers) { @@ -558,14 +558,26 @@ namespace MTSC.ServerSide } foreach (ClientData client in toRemove) { - foreach (IHandler handler in handlers) + try { - handler.ClientRemoved(this, client); + foreach (IHandler handler in handlers) + { + handler.ClientRemoved(this, client); + } + LogDebug("Client removed: " + client.TcpClient?.Client?.RemoteEndPoint?.ToString()); + client.Dispose(); } - LogDebug("Client removed: " + client.TcpClient?.Client?.RemoteEndPoint?.ToString()); - client.SslStream?.Dispose(); - client.TcpClient?.Dispose(); - Clients.Remove(client); + catch(Exception e) + { + foreach (IExceptionHandler exceptionHandler in exceptionHandlers) + { + if (exceptionHandler.HandleException(e)) + { + break; + } + } + } + clients.Remove(client); } toRemove.Clear(); } @@ -604,15 +616,7 @@ namespace MTSC.ServerSide client.LastMessageTime = DateTime.Now; LogDebug("Received message from " + client.TcpClient.Client.RemoteEndPoint.ToString() + "\nMessage length: " + message.MessageLength); - int retries = 0; - while(!InQueue.TryAdd((client, message))) - { - retries++; - if(retries > 5) - { - throw new QueueOperationException($"Failed to insert received message in {nameof(InQueue)}. Tried [{retries}] times"); - } - } + this._ProducerMessageInQueue.Enqueue((client, message)); } } catch(Exception e) @@ -689,19 +693,10 @@ namespace MTSC.ServerSide sslStream.AuthenticateAsServerAsync(this.certificate, this.RequestClientCertificate, this.SslProtocols, false).Wait(SslAuthenticationTimeout); } - Clients.Add(client); - Log("Accepted new connection: " + client.TcpClient.Client.RemoteEndPoint.ToString()); - foreach (IHandler handler in handlers) - { - if (handler.HandleClient(this, client)) - { - break; - } - } + this._ProducerClientQueue.Enqueue(client); } catch (Exception e) { - client.ToBeRemoved = true; foreach (IExceptionHandler exceptionHandler in exceptionHandlers) { if (exceptionHandler.HandleException(e)) @@ -709,6 +704,7 @@ namespace MTSC.ServerSide break; } } + client.Dispose(); } } #endregion