amacocian/MTSC
1
0
Fork 0
mirror of https://github.com/AlexMacocian/MTSC.git synced 2026-09-19 14:55:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix client reading loop for messages bigger than the buffer (#8)

Browse Source
This commit is contained in:
amacocian
2022-01-22 14:01:51 +01:00
committed by GitHub
parent 9d7a4ff75c
commit f8bf08d9ff
5 changed files with 41 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MTSC/CommunicationPrimitives.cs
+23 -11
View File
@@ -2,6 +2,7 @@
using MTSC.ServerSide;
using System;
using System.IO;
using System.Linq;
using System.Net.Security;
using System.Net.Sockets;
using System.Threading;
@@ -40,7 +41,7 @@ namespace MTSC
return new Message((uint)ms.Length, ms.ToArray());
}
public static async Task<Message> GetMessage(ClientData client, TimeSpan ReadTimeout)
public static async void LoopRead(ClientData client, Action<ClientData, Message> readMessageCallback)
{
Stream stream;
if (client.SslStream != null)
@@ -53,17 +54,28 @@ namespace MTSC
}
var buffer = new byte[1024];
var ms = new MemoryStream();
stream.ReadTimeout = (int)ReadTimeout.TotalMilliseconds;
int bytesRead;
do
while (true)
{
using var cts = new CancellationTokenSource();
cts.CancelAfter((int)ReadTimeout.TotalMilliseconds);
bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length, cts.Token);
ms.Write(buffer, 0, bytesRead);
} while (bytesRead > 0 && client.Socket.Available > 0);
return new Message((uint)ms.Length, ms.ToArray());
try
{
var bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length, client.CancellationToken);
var bytes = new byte[bytesRead];
if (bytesRead == 0)
{
client.ToBeRemoved = true;
return;
}
Array.Copy(buffer, 0, bytes, 0, bytesRead);
readMessageCallback(client, new Message((uint)bytes.Length, bytes));
(client as IActiveClient).UpdateLastReceivedMessage();
}
catch
{
client.ToBeRemoved = true;
return;
}
}
}
public static void SendMessage(Message message, Stream clientStream, SslStream sslStream = null)
MTSC/MTSC.csproj
+3 -3
View File
@@ -5,13 +5,13 @@
<TargetFrameworks>net48;netstandard2.0;netcoreapp3.1;net5.0;net6.0</TargetFrameworks>
<ApplicationIcon />
<StartupObject />
<Version>5.1</Version>
<Version>5.1.1</Version>
<LangVersion>latest</LangVersion>
<Authors>Alexandru-Victor Macocian</Authors>
<Product>MTSC</Product>
<Description>Modular TCP Server and Client</Description>
<AssemblyVersion>5.1.0.0</AssemblyVersion>
<FileVersion>5.1.0.0</FileVersion>
<AssemblyVersion>5.1.1.0</AssemblyVersion>
<FileVersion>5.1.1.0</FileVersion>
<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
<Platforms>AnyCPU;x64</Platforms>
<PackageProjectUrl>https://github.com/AlexMacocian/MTSC</PackageProjectUrl>
MTSC/ServerSide/ClientData.cs
-1
View File
@@ -48,7 +48,6 @@ namespace MTSC.ServerSide
public ResourceDictionary Resources { get; set; } = new ResourceDictionary();
IConsumerQueue<Message> IQueueHolder<Message>.ConsumerQueue => this.messageQueue;
bool IActiveClient.ReadingData { get; set; }
public ClientData(Socket socket)
{
MTSC/ServerSide/IActiveClient.cs
-4
View File
@@ -2,10 +2,6 @@
{
interface IActiveClient
{
/// <summary>
/// Indicates that there is currently a reading operation on the client
/// </summary>
bool ReadingData { get; set; }
/// <summary>
/// Updates the latest received message and activity time to DateTime.Now
/// </summary>
MTSC/ServerSide/Server.cs
+15 -31
View File
@@ -583,10 +583,6 @@ namespace MTSC.ServerSide
{
this.HandleException(e);
}
/*
* Check and gather messages from clients and place them in their queues.
*/
this.CheckAndGatherMessages();
/*
* Check if the server has any pending connections.
* If it has a new connection, process it.
@@ -607,7 +603,7 @@ namespace MTSC.ServerSide
/*
* Add all accepted clients to the list
*/
while(this.ConsumerClientQueue.TryDequeue(out var client))
while (this.ConsumerClientQueue.TryDequeue(out var client))
{
this.Log("Accepted new connection: " + client.Socket.RemoteEndPoint.ToString());
this.clients.Add(client);
@@ -618,6 +614,15 @@ namespace MTSC.ServerSide
break;
}
}
try
{
CommunicationPrimitives.LoopRead(client, this.MessageReceived);
}
catch (Exception e)
{
this.HandleException(e);
}
}
/*
@@ -775,34 +780,13 @@ namespace MTSC.ServerSide
this.HandleException(e);
}
}
private void CheckAndGatherMessages()
private void MessageReceived(ClientData client, Message message)
{
foreach(var client in this.Clients)
(client as IQueueHolder<Message>).Enqueue(message);
this.LogDebug($"Received message from {client.Socket.RemoteEndPoint as IPEndPoint} Message length: {message.MessageLength}");
if (this.LogMessageContents)
{
if (client.Socket.Available > 0 && !(client as IActiveClient).ReadingData)
{
(client as IActiveClient).ReadingData = true;
Task.Run(async () =>
{
try
{
var timeout = this.ReadTimeout;
var message = await CommunicationPrimitives.GetMessage(client, timeout);
(client as IQueueHolder<Message>).Enqueue(message);
this.LogDebug($"Received message from {client.Socket.RemoteEndPoint as IPEndPoint} Message length: {message.MessageLength}");
if (this.LogMessageContents)
{
this.LogDebug(Encoding.UTF8.GetString(message.MessageBytes));
}
(client as IActiveClient).ReadingData = false;
}
catch (Exception)
{
client.ToBeRemoved = true;
}
});
}
this.LogDebug(Encoding.UTF8.GetString(message.MessageBytes));
}
}
private void HandleClientMessages(ClientData client, IConsumerQueue<Message> messages)