amacocian/MTSC
1
0
Fork 0
mirror of https://github.com/AlexMacocian/MTSC.git synced 2026-07-15 14:59:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improvements to stream reading routine

Browse Source
This commit is contained in:
Alexandru Macocian
2021-09-30 18:42:18 +02:00
parent 21c793d708
commit 0b7c349a63
6 changed files with 30 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MTSC/CommunicationPrimitives.cs
+5 -3
View File
@@ -40,7 +40,7 @@ namespace MTSC
return new Message((uint)ms.Length, ms.ToArray());
}
public static Message GetMessage(ClientData client, TimeSpan ReadTimeout)
public static async Task<Message> GetMessage(ClientData client, TimeSpan ReadTimeout)
{
Stream stream;
if (client.SslStream != null)
@@ -58,9 +58,11 @@ namespace MTSC
int bytesRead;
do
{
bytesRead = stream.Read(buffer, 0, buffer.Length);
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);
} while (bytesRead > 0 && client.TcpClient.Available > 0);
return new Message((uint)ms.Length, ms.ToArray());
}
MTSC/MTSC.csproj
+3 -3
View File
@@ -5,13 +5,13 @@
<TargetFrameworks>netcoreapp2.1;net48;netstandard2.0;netcoreapp3.1;net5.0</TargetFrameworks>
<ApplicationIcon />
<StartupObject />
<Version>4.1.1</Version>
<Version>4.1.2</Version>
<LangVersion>latest</LangVersion>
<Authors>Alexandru-Victor Macocian</Authors>
<Product>MTSC</Product>
<Description>Modular TCP Server and Client</Description>
<AssemblyVersion>4.1.1.0</AssemblyVersion>
<FileVersion>4.1.1.0</FileVersion>
<AssemblyVersion>4.1.2.0</AssemblyVersion>
<FileVersion>4.1.2.0</FileVersion>
<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
<Platforms>AnyCPU;x64</Platforms>
<PackageProjectUrl>https://github.com/AlexMacocian/MTSC</PackageProjectUrl>
MTSC/ServerSide/Handlers/FtpHandler.cs
+1 -1
View File
@@ -22,7 +22,7 @@ namespace MTSC.ServerSide.Handlers
Initialized
}
private List<IFtpModule> ftpModules = new();
private readonly List<IFtpModule> ftpModules = new();
public TimeSpan ReadyDelay { get; set; } = TimeSpan.FromSeconds(1);
MTSC/ServerSide/Handlers/HttpHandler.cs
+4 -4
View File
@@ -102,11 +102,11 @@ namespace MTSC.ServerSide.Handlers
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;
HttpRequest request;
try
{
var trimmedMessageBytes = message.MessageBytes.TrimTrailingNullBytes();
byte[] messageBytes;
if (client.Resources.TryGetResource<FragmentedMessage>(out var fragmentedMessage))
{
var previousBytes = fragmentedMessage.Message;
@@ -147,7 +147,7 @@ namespace MTSC.ServerSide.Handlers
else
{
client.SetAffinity(this);
this.HandleIncompleteRequest(client, server, messageBytes, partialRequest);
this.HandleIncompleteRequest(client, server, messageBytes);
if (partialRequest != null && partialRequest.Headers.ContainsHeader(RequestHeaders.Expect) &&
partialRequest.Headers[RequestHeaders.Expect].Equals("100-continue", StringComparison.OrdinalIgnoreCase))
{
@@ -287,7 +287,7 @@ namespace MTSC.ServerSide.Handlers
}
#endregion
private void HandleIncompleteRequest(ClientData client, Server server, byte[] messageBytes, PartialHttpRequest partialRequest = null)
private void HandleIncompleteRequest(ClientData client, Server server, byte[] messageBytes)
{
client.Resources.SetResource(new FragmentedMessage() { Message = messageBytes, LastReceived = DateTime.Now });
server.LogDebug("Incomplete request received!");
MTSC/ServerSide/Handlers/WebsocketRoutingHandler.cs
+1 -1
View File
@@ -28,7 +28,7 @@ namespace MTSC.ServerSide.Handlers
Closed
}
#region Fields
private byte[] emptyData = new byte[0];
private readonly byte[] emptyData = new byte[0];
private DateTime ellapsedTime = DateTime.Now;
private DateTime previousHeartbeatProc = DateTime.Now;
private readonly Dictionary<string, (Type, Func<Server, HttpRequest, ClientData, RouteEnablerResponse>)> moduleDictionary =
MTSC/ServerSide/Server.cs
+16 -29
View File
@@ -36,7 +36,6 @@ namespace MTSC.ServerSide
private readonly List<IExceptionHandler> exceptionHandlers = new();
private readonly List<IServerUsageMonitor> serverUsageMonitors = new();
private readonly ProducerConsumerQueue<(ClientData, byte[])> messageOutQueue = new();
private readonly IServiceManager serviceManager = new ServiceManager();
#endregion
#region Private Properties
private IConsumerQueue<ClientData> ConsumerClientQueue { get => this.addQueue; }
@@ -103,7 +102,7 @@ namespace MTSC.ServerSide
/// <summary>
/// <see cref="IServiceManager"/> for configuring and retrieving services.
/// </summary>
public IServiceManager ServiceManager { get => this.serviceManager; }
public IServiceManager ServiceManager { get; } = new ServiceManager();
#endregion
#region Constructors
/// <summary>
@@ -162,7 +161,7 @@ namespace MTSC.ServerSide
where TService : TInterface
where TInterface : class
{
this.serviceManager.RegisterTransient<TInterface, TService>();
this.ServiceManager.RegisterTransient<TInterface, TService>();
return this;
}
/// <summary>
@@ -172,7 +171,7 @@ namespace MTSC.ServerSide
public Server AddTransientService<TService>()
where TService : class
{
this.serviceManager.RegisterTransient<TService>();
this.ServiceManager.RegisterTransient<TService>();
return this;
}
/// <summary>
@@ -183,7 +182,7 @@ namespace MTSC.ServerSide
where TService : TInterface
where TInterface : class
{
this.serviceManager.RegisterTransient<TInterface, TService>(serviceFactory);
this.ServiceManager.RegisterTransient<TInterface, TService>(serviceFactory);
return this;
}
/// <summary>
@@ -193,7 +192,7 @@ namespace MTSC.ServerSide
public Server AddTransientService<TService>(Func<Slim.IServiceProvider, TService> serviceFactory)
where TService : class
{
this.serviceManager.RegisterTransient<TService>(serviceFactory);
this.ServiceManager.RegisterTransient<TService>(serviceFactory);
return this;
}
/// <summary>
@@ -204,7 +203,7 @@ namespace MTSC.ServerSide
where TService : TInterface
where TInterface : class
{
this.serviceManager.RegisterSingleton<TInterface, TService>();
this.ServiceManager.RegisterSingleton<TInterface, TService>();
return this;
}
/// <summary>
@@ -214,7 +213,7 @@ namespace MTSC.ServerSide
public Server AddSingletonService<TService>()
where TService : class
{
this.serviceManager.RegisterSingleton<TService>();
this.ServiceManager.RegisterSingleton<TService>();
return this;
}
/// <summary>
@@ -225,7 +224,7 @@ namespace MTSC.ServerSide
where TService : TInterface
where TInterface : class
{
this.serviceManager.RegisterSingleton<TInterface, TService>(serviceFactory);
this.ServiceManager.RegisterSingleton<TInterface, TService>(serviceFactory);
return this;
}
/// <summary>
@@ -235,7 +234,7 @@ namespace MTSC.ServerSide
public Server AddSingletonService<TService>(Func<Slim.IServiceProvider, TService> serviceFactory)
where TService : class
{
this.serviceManager.RegisterSingleton<TService>(serviceFactory);
this.ServiceManager.RegisterSingleton<TService>(serviceFactory);
return this;
}
/// <summary>
@@ -406,7 +405,7 @@ namespace MTSC.ServerSide
public T GetService<T>()
where T : class
{
return this.serviceManager.GetService<T>();
return this.ServiceManager.GetService<T>();
}
/// <summary>
/// Get handler of provided type
@@ -528,8 +527,8 @@ namespace MTSC.ServerSide
toBeRunOnStartup.OnStartup(this);
}
this.serviceManager.RegisterServiceManager();
this.serviceManager.RegisterSingleton<Server, Server>(sp => this);
this.ServiceManager.RegisterServiceManager();
this.ServiceManager.RegisterSingleton<Server, Server>(sp => this);
DateTime startLoopTime;
while (this.running)
{
@@ -666,13 +665,6 @@ namespace MTSC.ServerSide
while (this.ConsumerMessageOutQueue.TryDequeue(out var tuple))
{
(var client, var bytes) = tuple;
if (client.TcpClient.Available > 0)
{
/*
* Don't send while client is still sending
*/
continue;
}
try
{
@@ -744,19 +736,14 @@ namespace MTSC.ServerSide
if (client.TcpClient.Available > 0 && !(client as IActiveClient).ReadingData)
{
(client as IActiveClient).ReadingData = true;
Task.Run(() =>
Task.Run(async () =>
{
try
{
var timeout = this.ReadTimeout;
if (client.TcpClient.Available < 1000)
{
timeout = TimeSpan.FromMilliseconds(50);
}
var message = CommunicationPrimitives.GetMessage(client, timeout);
var message = await CommunicationPrimitives.GetMessage(client, timeout);
(client as IQueueHolder<Message>).Enqueue(message);
this.LogDebug($"Received message from {(client.TcpClient.Client.RemoteEndPoint as IPEndPoint)} Message length: {message.MessageLength}");
this.LogDebug($"Received message from {client.TcpClient.Client.RemoteEndPoint as IPEndPoint} Message length: {message.MessageLength}");
if (this.LogMessageContents)
{
this.LogDebug(Encoding.UTF8.GetString(message.MessageBytes));
@@ -852,7 +839,7 @@ namespace MTSC.ServerSide
this.EncryptionPolicy);
client.SslStream = sslStream;
if(sslStream.AuthenticateAsServerAsync(this.certificate, this.RequestClientCertificate, this.SslProtocols, false).Wait(this.SslAuthenticationTimeout))
if (sslStream.AuthenticateAsServerAsync(this.certificate, this.RequestClientCertificate, this.SslProtocols, false).Wait(this.SslAuthenticationTimeout))
{
/*
* Client authenticated in the alloted time