- Fixed messages being handled non-sequentially during parallel processing of the clients

- Made unit tests use SSL encryption
- Added netcore3.1 to build list

MTSC.UnitTests/E2ETests.cs

@@ -34,7 +34,10 @@ namespace MTSC.UnitTests
         [ClassInitialize]
         public static void InitializeServer(TestContext testContext)
         {
+            ServicePointManager.ServerCertificateValidationCallback = (_, __, ___, ____) => true;
+
             Server = new ServerSide.Server(800)
+                .WithCertificate(new X509Certificate2("myCert.cer"))
                 .AddHandler(new WebsocketRoutingHandler()
                     .AddRoute("echo", new EchoWebsocketModule()
                         .WithReceiveTemplateProvider((message) => UTF8Encoding.UTF8.GetString(message.Data))
@@ -49,6 +52,7 @@ namespace MTSC.UnitTests
                     .AddRoute(HttpMessage.HttpMethods.Get, "", new Http200Module())
                     .AddRoute(HttpMessage.HttpMethods.Get, "query", new TestQueryModule())
                     .AddRoute(HttpMessage.HttpMethods.Get, "echo", new EchoModule())
+                    .AddRoute(HttpMessage.HttpMethods.Post, "echo", new EchoModule())
                     .AddRoute(HttpMessage.HttpMethods.Get, "long-running", new LongRunningModule())
                     .WithFragmentsExpirationTime(TimeSpan.FromMilliseconds(500))
                     .WithMaximumSize(300))
@@ -63,11 +67,11 @@ namespace MTSC.UnitTests
         public async Task ServerRespondsDuringLongRunningTask()
         {
             HttpClient httpClient = new HttpClient();
-            httpClient.BaseAddress = new Uri("http://localhost:800");
+            httpClient.BaseAddress = new Uri("https://localhost:800");
             var longRunningTask = httpClient.GetAsync("long-running");
             int responses = 0;
             HttpClient client2 = new HttpClient();
-            client2.BaseAddress = new Uri("http://localhost:800");
+            client2.BaseAddress = new Uri("https://localhost:800");
             while (!longRunningTask.IsCompleted)
             {
                 var echoResponse = await client2.GetAsync("echo");
@@ -82,7 +86,7 @@ namespace MTSC.UnitTests
         public void HelloWorldHTTP()
         {
             HttpClient httpClient = new HttpClient();
-            httpClient.BaseAddress = new Uri("http://localhost:800");
+            httpClient.BaseAddress = new Uri("https://localhost:800");
             var result = httpClient.GetAsync("").Result;
             Assert.AreEqual(result.StatusCode, System.Net.HttpStatusCode.OK);
         }
@@ -96,6 +100,7 @@ namespace MTSC.UnitTests
             client.SetServerAddress("127.0.0.1")
                 .SetPort(800)
                 .AddHandler(notifyHandler)
+                .WithSsl(true)
                 .Connect();
 
             HttpRequest request = new HttpRequest();
@@ -120,6 +125,8 @@ namespace MTSC.UnitTests
                 }
             }
             HttpResponse response = HttpResponse.FromBytes(receivedMessage);
+            //Trim the null bytes from encryption/decryption
+            response.BodyString = response.BodyString.Trim('\0');
             Assert.AreEqual(response.StatusCode, HttpMessage.StatusCodes.OK);
             Assert.AreEqual(response.BodyString, "Brought a message to you my guy!");
         }
@@ -133,6 +140,7 @@ namespace MTSC.UnitTests
             client.SetServerAddress("127.0.0.1")
                 .SetPort(800)
                 .AddHandler(notifyHandler)
+                .WithSsl(true)
                 .Connect();
 
             HttpRequest request = new HttpRequest();
@@ -168,6 +176,7 @@ namespace MTSC.UnitTests
             client.SetServerAddress("127.0.0.1")
                 .SetPort(800)
                 .AddHandler(notifyHandler)
+                .WithSsl(true)
                 .Connect();
 
             HttpRequest request = new HttpRequest();
@@ -199,7 +208,7 @@ namespace MTSC.UnitTests
         [TestMethod]
         public void GetWithQueryHttp()
         {
-            var builder = new UriBuilder("http://localhost:800/query");
+            var builder = new UriBuilder("https://localhost:800/query");
             var query = HttpUtility.ParseQueryString(builder.Query);
             query["key1"] = "value1";
             query["key2"] = "value2";
@@ -217,7 +226,7 @@ namespace MTSC.UnitTests
         {
             byte[] bytes = new byte[100];
             ClientWebSocket client = new ClientWebSocket();
-            client.ConnectAsync(new Uri("ws://localhost:800/echo"), CancellationToken.None).Wait();
+            client.ConnectAsync(new Uri("wss://localhost:800/echo"), CancellationToken.None).Wait();
             client.SendAsync(ASCIIEncoding.ASCII.GetBytes("Hello world!"), WebSocketMessageType.Text, true, CancellationToken.None).Wait();
             client.ReceiveAsync(bytes, CancellationToken.None).Wait();
             var resultString = ASCIIEncoding.ASCII.GetString(bytes, 0, 12);
@@ -228,7 +237,7 @@ namespace MTSC.UnitTests
         public void HTTPStressTest()
         {
             var httpClient = new HttpClient();
-            httpClient.BaseAddress = new Uri("http://localhost:800");
+            httpClient.BaseAddress = new Uri("https://localhost:800");
             for(int i = 0; i < stressIterations; i++)
             {
                 var startTime = DateTime.Now;

MTSC.UnitTests/MTSC.UnitTests.csproj

@@ -18,6 +18,9 @@
   </ItemGroup>
 
   <ItemGroup>
+    <None Update="myCert.cer">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </None>
     <None Update="powershellcert.pfx">
       <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
     </None>

MTSC/ClientSide/Client.cs

@@ -51,6 +51,11 @@ namespace MTSC.Client
         }
         #endregion
         #region Public Methods
+        public Client WithSsl(bool ssl)
+        {
+            this.useSsl = ssl;
+            return this;
+        }
         /// <summary>
         /// Add a message to the message queue.
         /// </summary>

MTSC/Common/Http/PartialHttpRequest.cs

@@ -407,10 +407,6 @@ namespace MTSC.Common.Http
                     this.Body = ms.ReadRemainingBytes();
                 }
             }
-            /*
-             * Trim all trailing null characters left over from SSL encryption.
-             */
-            this.BodyString = this.BodyString.Trim('\0');
             Complete = true;
             return;
         }

MTSC/MTSC.csproj

@@ -2,15 +2,15 @@
 
   <PropertyGroup>
     <OutputType>Library</OutputType>
-    <TargetFrameworks>netcoreapp2.1;net48;netstandard2.0;netcoreapp3.0</TargetFrameworks>
+    <TargetFrameworks>netcoreapp2.1;net48;netstandard2.0;netcoreapp3.0;netcoreapp3.1</TargetFrameworks>
     <ApplicationIcon />
     <StartupObject />
-    <Version>2.4.1</Version>
+    <Version>2.4.2</Version>
     <Authors>Alexandru-Victor Macocian</Authors>
     <Product>MTSC</Product>
     <Description>Modular TCP Server and Client</Description>
-    <AssemblyVersion>0.2.4.1</AssemblyVersion>
-    <FileVersion>0.2.4.1</FileVersion>
+    <AssemblyVersion>0.2.4.2</AssemblyVersion>
+    <FileVersion>0.2.4.2</FileVersion>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
     <Platforms>AnyCPU;x64</Platforms>
     <PackageProjectUrl>https://github.com/AlexMacocian/MTSC</PackageProjectUrl>

MTSC/ServerSide/ClientData.cs

@@ -1,4 +1,5 @@
-using System;
+using MTSC.Common;
+using System;
 using System.Net.Security;
 using System.Net.Sockets;
 
@@ -7,8 +8,10 @@ namespace MTSC.ServerSide
     /// <summary>
     /// Structure containing client information.
     /// </summary>
-    public class ClientData : IDisposable, IActiveClient
+    public class ClientData : IDisposable, IActiveClient, IQueueHolder<Message>
     {
+        private ProducerConsumerQueue<Message> messageQueue = new ProducerConsumerQueue<Message>();
+
         public TcpClient TcpClient;
         /// <summary>
         /// Latest datetime when a message has been received from the client
@@ -18,6 +21,9 @@ namespace MTSC.ServerSide
         /// Latest datetime when a message has been received or sent to the client
         /// </summary>
         public DateTime LastActivityTime { get; private set; } = DateTime.Now;
+
+        IConsumerQueue<Message> IQueueHolder<Message>.ConsumerQueue => messageQueue;
+
         public bool ToBeRemoved = false;
         public SslStream SslStream = null;
         public ResourceDictionary Resources = new ResourceDictionary();
@@ -37,6 +43,21 @@ namespace MTSC.ServerSide
             LastActivityTime = DateTime.Now;
         }
 
+        void IQueueHolder<Message>.Enqueue(Message value)
+        {
+            (messageQueue as IProducerQueue<Message>).Enqueue(value);
+        }
+
+        Message IQueueHolder<Message>.Dequeue()
+        {
+            return (messageQueue as IConsumerQueue<Message>).Dequeue();
+        }
+
+        bool IQueueHolder<Message>.TryDequeue(out Message Value)
+        {
+            return (messageQueue as IConsumerQueue<Message>).TryDequeue(out Value);
+        }
+
         #region IDisposable Support
         private bool disposedValue = false; // To detect redundant calls
 

MTSC/ServerSide/Handlers/HttpHandler.cs

@@ -96,30 +96,31 @@ namespace MTSC.ServerSide.Handlers
             byte[] messageBytes = null;
             try
             {
+                var trimmedMessageBytes = message.MessageBytes.TrimTrailingNullBytes();
                 if (fragmentedMessages.ContainsKey(client))
                 {
                     byte[] previousBytes = fragmentedMessages[client].Item1;
-                    if(previousBytes.Length + message.MessageBytes.Length > MaximumRequestSize)
+                    if(previousBytes.Length + trimmedMessageBytes.Length > MaximumRequestSize)
                     {
                         // Discard the message if it is too big
-                        server.LogDebug($"Discarded message. Message size [{previousBytes.Length + message.MessageBytes.Length}] > [{MaximumRequestSize}]");
+                        server.LogDebug($"Discarded message. Message size [{previousBytes.Length + trimmedMessageBytes.Length}] > [{MaximumRequestSize}]");
                         fragmentedMessages.TryRemove(client, out _);
                         return false;
                     }
-                    byte[] repackagingBuffer = new byte[previousBytes.Length + message.MessageBytes.Length];
+                    byte[] repackagingBuffer = new byte[previousBytes.Length + trimmedMessageBytes.Length];
                     Array.Copy(previousBytes, 0, repackagingBuffer, 0, previousBytes.Length);
-                    Array.Copy(message.MessageBytes, 0, repackagingBuffer, previousBytes.Length, message.MessageBytes.Length);
+                    Array.Copy(trimmedMessageBytes, 0, repackagingBuffer, previousBytes.Length, trimmedMessageBytes.Length);
                     messageBytes = repackagingBuffer;
                 }
                 else
                 {
-                    if(message.MessageBytes.Length > MaximumRequestSize)
+                    if(trimmedMessageBytes.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 = trimmedMessageBytes;
                 }
                 messageBytes = messageBytes.TrimTrailingNullBytes();
                 var partialRequest = PartialHttpRequest.FromBytes(messageBytes);

MTSC/ServerSide/Handlers/HttpRoutingHandler.cs

@@ -93,32 +93,32 @@ namespace MTSC.ServerSide.Handlers
             byte[] messageBytes = null;
             try
             {
+                var trimmedMessageBytes = message.MessageBytes.TrimTrailingNullBytes();
                 if (fragmentedMessages.ContainsKey(client))
                 {
                     byte[] previousBytes = fragmentedMessages[client].Item1;
-                    if (previousBytes.Length + message.MessageBytes.Length > MaximumRequestSize)
+                    if (previousBytes.Length + trimmedMessageBytes.Length > MaximumRequestSize)
                     {
                         // Discard the message if it is too big
-                        server.LogDebug($"Discarded message. Message size [{previousBytes.Length + message.MessageBytes.Length}] > [{MaximumRequestSize}]");
+                        server.LogDebug($"Discarded message. Message size [{previousBytes.Length + trimmedMessageBytes.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);
+                    Array.Copy(trimmedMessageBytes, 0, repackagingBuffer, previousBytes.Length, trimmedMessageBytes.Length);
                     messageBytes = repackagingBuffer;
                 }
                 else
                 {
-                    if (message.MessageBytes.Length > MaximumRequestSize)
+                    if (trimmedMessageBytes.Length > MaximumRequestSize)
                     {
                         // Discard the message if it is too big
-                        server.LogDebug($"Discarded message. Message size [{message.MessageBytes.Length}] > [{MaximumRequestSize}]");
+                        server.LogDebug($"Discarded message. Message size [{trimmedMessageBytes.Length}] > [{MaximumRequestSize}]");
                         return false;
                     }
-                    messageBytes = message.MessageBytes;
+                    messageBytes = trimmedMessageBytes;
                 }
-                messageBytes = messageBytes.TrimTrailingNullBytes();
                 var partialRequest = PartialHttpRequest.FromBytes(messageBytes);
                 if (partialRequest.Complete)
                     request = partialRequest.ToRequest();

MTSC/ServerSide/IQueueHolder.cs

@@ -0,0 +1,12 @@
+using MTSC.Common;
+
+namespace MTSC.ServerSide
+{
+    interface IQueueHolder<T>
+    {
+        IConsumerQueue<T> ConsumerQueue { get; }
+        void Enqueue(T value);
+        T Dequeue();
+        bool TryDequeue(out T Value);
+    }
+}

MTSC/ServerSide/Schedulers/FireTasksAndForgetScheduler.cs

@@ -1,17 +1,18 @@
 using MTSC.Common;
 using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace MTSC.ServerSide.Schedulers
 {
     public class FireTasksAndForgetScheduler : IScheduler
     {
-        public void ScheduleHandling(IConsumerQueue<(ClientData, Message)> inQueue, Action<ClientData, Message> messageHandlingProcedure)
+        void IScheduler.ScheduleHandling(List<(ClientData, IConsumerQueue<Message>)> clientsQueues, Action<ClientData, IConsumerQueue<Message>> messageHandlingProcedure)
         {
-            while(inQueue.TryDequeue(out var tuple))
+            foreach(var tuple in clientsQueues)
             {
-                (var client, var message) = tuple;
-                Task.Run(() => { messageHandlingProcedure.Invoke(client, message); });
+                (var client, var messageQueue) = tuple;
+                Task.Run(() => messageHandlingProcedure.Invoke(client, messageQueue));
             }
         }
     }

MTSC/ServerSide/Schedulers/IScheduler.cs

@@ -1,10 +1,11 @@
 using MTSC.Common;
 using System;
+using System.Collections.Generic;
 
 namespace MTSC.ServerSide.Schedulers
 {
     public interface IScheduler
     {
-        void ScheduleHandling(IConsumerQueue<(ClientData, Message)> inQueue, Action<ClientData, Message> messageHandlingProcedure);
+        void ScheduleHandling(List<(ClientData, IConsumerQueue<Message>)> clientsQueues, Action<ClientData, IConsumerQueue<Message>> messageHandlingProcedure);
     }
 }

MTSC/ServerSide/Schedulers/ParallelScheduler.cs

@@ -7,16 +7,15 @@ namespace MTSC.ServerSide.Schedulers
 {
     public class ParallelScheduler : IScheduler
     {
-        void IScheduler.ScheduleHandling(IConsumerQueue<(ClientData, Message)> inQueue, Action<ClientData, Message> messageHandlingProcedure)
+        void IScheduler.ScheduleHandling(List<(ClientData, IConsumerQueue<Message>)> clientsQueues, Action<ClientData, IConsumerQueue<Message>> messageHandlingProcedure)
         {
             List<Action> actionList = new List<Action>();
 
-            while(inQueue.TryDequeue(out var tuple))
+            foreach(var tuple in clientsQueues)
             {
-                (var client, var message) = tuple;
-                actionList.Add(new Action(() => { messageHandlingProcedure.Invoke(client, message); }));
+                (var client, var messageQueue) = tuple;
+                actionList.Add(new Action(() => messageHandlingProcedure.Invoke(client, messageQueue)));
             }
-            Parallel.Invoke(actionList.ToArray());
         }
     }
 }

MTSC/ServerSide/Schedulers/SequentialProcessingScheduler.cs

@@ -1,17 +1,18 @@
 using MTSC.Common;
 using System;
 using System.Collections.Concurrent;
+using System.Collections.Generic;
 
 namespace MTSC.ServerSide.Schedulers
 {
     public class SequentialProcessingScheduler : IScheduler
     {
-        void IScheduler.ScheduleHandling(IConsumerQueue<(ClientData, Message)> inQueue, Action<ClientData, Message> messageHandlingProcedure)
+        void IScheduler.ScheduleHandling(List<(ClientData, IConsumerQueue<Message>)> clientsQueues, Action<ClientData, IConsumerQueue<Message>> messageHandlingProcedure)
         {
-            while (inQueue.TryDequeue(out var tuple))
+            foreach(var tuple in clientsQueues)
             {
-                (var client, var message) = tuple;
-                messageHandlingProcedure.Invoke(client, message);
+                (var client, var messageQueue) = tuple;
+                messageHandlingProcedure.Invoke(client, messageQueue);
             }
         }
     }

MTSC/ServerSide/Schedulers/TaskAwaiterScheduler.cs

@@ -7,13 +7,13 @@ namespace MTSC.ServerSide.Schedulers
 {
     public class TaskAwaiterScheduler : IScheduler
     {
-        void IScheduler.ScheduleHandling(IConsumerQueue<(ClientData, Message)> inQueue, Action<ClientData, Message> messageHandlingProcedure)
+        void IScheduler.ScheduleHandling(List<(ClientData, IConsumerQueue<Message>)> clientsQueues, Action<ClientData, IConsumerQueue<Message>> messageHandlingProcedure)
         {
             List<Task> tasks = new List<Task>();
-            while(inQueue.TryDequeue(out var tuple))
+            foreach(var tuple in clientsQueues)
             {
-                (var client, var message) = tuple;
-                tasks.Add(Task.Run(() => messageHandlingProcedure.Invoke(client, message)));
+                (var client, var messageQueue) = tuple;
+                tasks.Add(Task.Run(() => messageHandlingProcedure.Invoke(client, messageQueue)));
             }
             Task.WaitAll(tasks.ToArray());
         }

MTSC/ServerSide/Server.cs

@@ -8,6 +8,7 @@ using MTSC.ServerSide.UsageMonitors;
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Linq;
 using System.Net;
 using System.Net.Security;
 using System.Net.Sockets;
@@ -33,15 +34,12 @@ namespace MTSC.ServerSide
         List<ILogger> loggers = new List<ILogger>();
         List<IExceptionHandler> exceptionHandlers = new List<IExceptionHandler>();
         List<IServerUsageMonitor> serverUsageMonitors = new List<IServerUsageMonitor>();
-        ProducerConsumerQueue<(ClientData, Message)> messageInQueue = new ProducerConsumerQueue<(ClientData, Message)>();
         ProducerConsumerQueue<(ClientData, byte[])> messageOutQueue = new ProducerConsumerQueue<(ClientData, byte[])>();
         #endregion
         #region Private Properties
         private IConsumerQueue<ClientData> _ConsumerClientQueue { get => addQueue; }
         private IProducerQueue<ClientData> _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
         /// <summary>
@@ -444,7 +442,11 @@ namespace MTSC.ServerSide
                  * Call the scheduler to handle all received messages and distribute them to the handlers
                  */
 
-                Scheduler.ScheduleHandling(this._ConsumerMessageInQueue, HandleClientMessage);
+                Scheduler.ScheduleHandling(
+                    clients
+                        .Select(client => (client, (client as IQueueHolder<Message>).ConsumerQueue))
+                        .ToList(),
+                    HandleClientMessages);
 
                 /*
                  * Iterate through all the handlers, running periodic operations.
@@ -617,7 +619,7 @@ namespace MTSC.ServerSide
                         (client as IActiveClient).UpdateLastReceivedMessage();
                         LogDebug("Received message from " + client.TcpClient.Client.RemoteEndPoint.ToString() +
                                 "\nMessage length: " + message.MessageLength);
-                        this._ProducerMessageInQueue.Enqueue((client, message));
+                        (client as IQueueHolder<Message>).Enqueue(message);
                     }
                 }
                 catch(Exception e)
@@ -633,6 +635,14 @@ namespace MTSC.ServerSide
             }
         }
 
+        private void HandleClientMessages(ClientData client, IConsumerQueue<Message> messages)
+        {
+            while(messages.TryDequeue(out var message))
+            {
+                HandleClientMessage(client, message);
+            }
+        }
+
         private void HandleClientMessage(ClientData client, Message message)
         {
             foreach (IHandler handler in handlers)