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

Setup for multiple listener types

Browse Source
This commit is contained in:
Alexandru Macocian
2021-10-01 17:25:32 +02:00
parent 032edb81d6
commit 8eb0fe06c0
16 changed files with 212 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MTSC-TestClient/Program.cs
-3
View File
@@ -2,7 +2,6 @@
using MTSC.Client.Handlers;
using MTSC.ClientSide;
using MTSC.Common.WebSockets.ClientModules;
using MTSC.Logging;
using System;
namespace MTSC_TestClient
@@ -21,8 +20,6 @@ namespace MTSC_TestClient
.AddHandler(websocketHandler.AddModule(chatModule))
//.AddHandler(new EncryptionHandler())
//.AddHandler(new BroadcastHandler())
.AddLogger(new ConsoleLogger())
.AddLogger(new DebugConsoleLogger())
.Connect();
while (true)
{
MTSC-TestServer/Program.cs
-7
View File
@@ -1,14 +1,9 @@
using MTSC.Common.Ftp.FtpModules;
using MTSC.Common.Http.RoutingModules;
using MTSC.Common.Http.ServerModules;
using MTSC.Common.WebSockets.ServerModules;
using MTSC.Exceptions;
using MTSC.Logging;
using MTSC.ServerSide;
using MTSC.ServerSide.Handlers;
using MTSC.ServerSide.UsageMonitors;
using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
namespace MTSC_TestServer
{
@@ -18,8 +13,6 @@ namespace MTSC_TestServer
{
var server = new Server(800);
server
.AddLogger(new ConsoleLogger())
.AddLogger(new DebugConsoleLogger())
.AddServerUsageMonitor(new TickrateEnforcer()
.SetTicksPerSecond(60)
.SetSilent(true))
MTSC.UnitTests/ConsoleLogger.cs
+24
View File
@@ -0,0 +1,24 @@
using System;
using Microsoft.Extensions.Logging;
namespace MTSC.UnitTests
{
public sealed class ConsoleLogger : ILogger
{
public IDisposable BeginScope<TState>(TState state)
{
throw new NotImplementedException();
}
public bool IsEnabled(LogLevel logLevel)
{
return true;
}
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func<TState, Exception, string> formatter)
{
var message = formatter(state, exception);
Console.WriteLine(message);
}
}
}
MTSC.UnitTests/E2ETests.cs
+45 -6
View File
@@ -1,3 +1,4 @@
using Microsoft.Extensions.Logging;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MTSC.Common.Http;
using MTSC.Common.Http.RoutingModules;
@@ -7,6 +8,7 @@ using MTSC.ServerSide.Schedulers;
using MTSC.UnitTests.RoutingModules;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
@@ -25,11 +27,11 @@ namespace MTSC.UnitTests
public class E2ETests
{
private volatile byte[] receivedMessage = null;
private static int stressIterations = 100000;
private static int stressIterations = 100;
public TestContext TestContext { get; set; }
static ServerSide.Server Server { get; set; }
[ClassInitialize]
[AssemblyInitialize]
public static void InitializeServer(TestContext testContext)
{
ServicePointManager.ServerCertificateValidationCallback += (s, e, o, p) => true;
@@ -39,7 +41,7 @@ namespace MTSC.UnitTests
.AddHandler(new WebsocketRoutingHandler()
.AddRoute<EchoWebsocketModule>("echo")
.AddRoute<EchoWebsocketModule2>("echo2")
.AddRoute<RoutingModules.HelloWorldModule>("hello-world")
.AddRoute<HelloWorldModule>("hello-world")
.WithHeartbeatEnabled(true)
.WithHeartbeatFrequency(TimeSpan.FromMilliseconds(100)))
.AddHandler(new HttpRoutingHandler()
@@ -56,6 +58,7 @@ namespace MTSC.UnitTests
.AddExceptionHandler(new ExceptionConsoleLogger())
.SetScheduler(new ParallelScheduler())
.WithSslAuthenticationTimeout(TimeSpan.FromMilliseconds(100));
Server.ServiceManager.RegisterSingleton<ILogger, ConsoleLogger>();
Server.RunAsync();
}
@@ -302,11 +305,13 @@ namespace MTSC.UnitTests
[DataRow("echo2")]
public async Task EchoWebsocket(string endpoint)
{
using var cts = new CancellationTokenSource();
cts.CancelAfter(15000);
var bytes = new byte[100];
var client = new ClientWebSocket();
await client.ConnectAsync(new Uri($"ws://localhost:800/{endpoint}"), CancellationToken.None);
await client.SendAsync(Encoding.ASCII.GetBytes("Hello world!"), WebSocketMessageType.Text, true, CancellationToken.None);
await client.ReceiveAsync(bytes, CancellationToken.None);
await client.ConnectAsync(new Uri($"ws://localhost:800/{endpoint}"), cts.Token);
await client.SendAsync(Encoding.ASCII.GetBytes("Hello world!"), WebSocketMessageType.Text, true, cts.Token);
await client.ReceiveAsync(bytes, cts.Token);
var resultString = Encoding.ASCII.GetString(bytes, 0, 12);
Assert.AreEqual(resultString, "Hello world!");
}
@@ -330,6 +335,40 @@ namespace MTSC.UnitTests
Assert.AreEqual(resultString, "Not hello world!");
}
[TestMethod]
public async Task StressTest()
{
for (var round = 0; round < 10; round++)
{
var clients = new List<HttpClient>();
for (var i = 0; i < 100; i++)
{
clients.Add(new HttpClient { BaseAddress = new Uri("http://localhost:800") });
}
for (var i = 0; i < stressIterations; i++)
{
Console.WriteLine($"Memory: {Process.GetCurrentProcess().PrivateMemorySize64}");
var tasks = new List<Task<HttpResponseMessage>>();
foreach (var client in clients)
{
tasks.Add(client.GetAsync(""));
}
Task.WaitAll(tasks.ToArray());
foreach (var task in tasks)
{
task.Result.EnsureSuccessStatusCode();
}
}
foreach (var client in clients)
{
client.Dispose();
}
}
}
[ClassCleanup]
public static void CleanupServer()
{
MTSC/ClientSide/Client.cs
+2 -2
View File
@@ -222,7 +222,7 @@ namespace MTSC.Client
var ipEndpoint = new IPEndPoint(ipAddress, this.Port);
this.tcpClient.Connect(ipEndpoint);
this.safeNetworkStream = new TimeoutSuppressedStream(this.tcpClient);
this.safeNetworkStream = new TimeoutSuppressedStream(this.tcpClient.GetStream());
if (shouldUseSsl)
{
this.CertificateValidationCallback = this.CertificateValidationCallback ?? new RemoteCertificateValidationCallback(ValidateServerCertificate);
@@ -346,7 +346,7 @@ namespace MTSC.Client
handler.HandleSendMessage(this, ref sendMessage);
}
CommunicationPrimitives.SendMessage(this.tcpClient, sendMessage, this.sslStream);
CommunicationPrimitives.SendMessage(sendMessage, this.safeNetworkStream, this.sslStream);
}
}
MTSC/Common/Ftp/FtpModules/DataConnectionModule.cs
+1 -1
View File
@@ -67,7 +67,7 @@ namespace MTSC.Common.Ftp.FtpModules
ftpData.TransferDetails.Mode = TransferDetails.TransferMode.Passive;
ftpData.TransferDetails.Socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
ftpData.TransferDetails.Socket.Bind(new IPEndPoint(IPAddress.Any, 0));
var address = ((IPEndPoint)client.TcpClient.Client.LocalEndPoint).Address.GetAddressBytes();
var address = ((IPEndPoint)client.Socket.LocalEndPoint).Address.GetAddressBytes();
var portBytes = BitConverter.GetBytes((short)ftpData.TransferDetails.LocalDataPort);
if (BitConverter.IsLittleEndian)
MTSC/Common/TimeoutSuppressedStream.cs
+12 -6
View File
@@ -10,9 +10,9 @@ namespace MTSC.Common
{
private readonly NetworkStream innerStream;
public TimeoutSuppressedStream(TcpClient tcpClient)
public TimeoutSuppressedStream(NetworkStream networkStream)
{
this.innerStream = tcpClient.GetStream();
this.innerStream = networkStream;
}
public override int Read(byte[] buffer, int offset, int count)
@@ -35,8 +35,6 @@ namespace MTSC.Common
throw;
}
}
public override bool CanRead => this.innerStream.CanRead;
public override bool CanSeek => this.innerStream.CanSeek;
public override bool CanTimeout => this.innerStream.CanTimeout;
@@ -53,18 +51,26 @@ namespace MTSC.Common
public override void SetLength(long value) => this.innerStream.SetLength(value);
public override void Write(byte[] buffer, int offset, int count) => this.innerStream.Write(buffer, offset, count);
protected override void Dispose(bool disposing)
{
if (disposing)
{
this.innerStream.Dispose();
}
base.Dispose(disposing);
}
public override long Position
{
get { return this.innerStream.Position; }
set { this.innerStream.Position = value; }
}
public override int ReadTimeout
{
get { return this.innerStream.ReadTimeout; }
set { this.innerStream.ReadTimeout = value; }
}
public override int WriteTimeout
{
get { return this.innerStream.WriteTimeout; }
MTSC/CommunicationPrimitives.cs
+3 -3
View File
@@ -62,11 +62,11 @@ namespace MTSC
cts.CancelAfter((int)ReadTimeout.TotalMilliseconds);
bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length, cts.Token);
ms.Write(buffer, 0, bytesRead);
} while (bytesRead > 0 && client.TcpClient.Available > 0);
} while (bytesRead > 0 && client.Socket.Available > 0);
return new Message((uint)ms.Length, ms.ToArray());
}
public static void SendMessage(TcpClient client, Message message, SslStream sslStream = null)
public static void SendMessage(Message message, Stream clientStream, SslStream sslStream = null)
{
Stream stream;
if (sslStream != null)
@@ -75,7 +75,7 @@ namespace MTSC
}
else
{
stream = client.GetStream();
stream = clientStream;
}
stream.Write(message.MessageBytes, 0, (int)message.MessageLength);
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.2.0</Version>
<Version>4.3.0</Version>
<LangVersion>latest</LangVersion>
<Authors>Alexandru-Victor Macocian</Authors>
<Product>MTSC</Product>
<Description>Modular TCP Server and Client</Description>
<AssemblyVersion>4.2.0.0</AssemblyVersion>
<FileVersion>4.2.0.0</FileVersion>
<AssemblyVersion>4.3.0.0</AssemblyVersion>
<FileVersion>4.3.0.0</FileVersion>
<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
<Platforms>AnyCPU;x64</Platforms>
<PackageProjectUrl>https://github.com/AlexMacocian/MTSC</PackageProjectUrl>
MTSC/ServerSide/ClientData.cs
+6 -5
View File
@@ -13,7 +13,7 @@ namespace MTSC.ServerSide
{
private readonly ProducerConsumerQueue<Message> messageQueue = new();
public TcpClient TcpClient { get; }
public Socket Socket { get; }
/// <summary>
/// Latest datetime when a message has been received from the client
/// </summary>
@@ -46,10 +46,11 @@ namespace MTSC.ServerSide
IConsumerQueue<Message> IQueueHolder<Message>.ConsumerQueue => this.messageQueue;
bool IActiveClient.ReadingData { get; set; }
public ClientData(TcpClient client)
public ClientData(Socket socket)
{
this.TcpClient = client;
this.SafeNetworkStream = new TimeoutSuppressedStream(this.TcpClient);
this.Socket = socket;
var ns = new NetworkStream(this.Socket);
this.SafeNetworkStream = new TimeoutSuppressedStream(ns);
}
/// <summary>
/// Sets the affinity of the client.
@@ -117,7 +118,7 @@ namespace MTSC.ServerSide
// TODO: dispose managed state (managed objects).
}
this.TcpClient?.Dispose();
this.SafeNetworkStream?.Dispose();
this.SslStream?.Dispose();
this.Resources?.Dispose();
MTSC/ServerSide/Handlers/BroadcastHandler.cs
+1 -1
View File
@@ -25,7 +25,7 @@ namespace MTSC.ServerSide.Handlers
bool IHandler.HandleReceivedMessage(Server server, ClientData client, Message message)
{
server.LogDebug("Broadcast: " + UnicodeEncoding.Unicode.GetString(message.MessageBytes));
server.LogDebug("From: " + client.TcpClient.Client.RemoteEndPoint.ToString());
server.LogDebug("From: " + client.Socket.RemoteEndPoint.ToString());
foreach(var clientStruct in server.Clients)
{
server.QueueMessage(clientStruct, message.MessageBytes);
MTSC/ServerSide/Handlers/WebsocketHandler.cs
+1 -1
View File
@@ -117,7 +117,7 @@ namespace MTSC.ServerSide.Handlers
response.Headers[WebsocketHeaderAcceptKey] = returnBase64Key;
server.QueueMessage(client, response.GetPackedResponse(false));
client.Resources.SetResource(SocketState.Established);
server.LogDebug("Websocket initialized " + client.TcpClient.Client.RemoteEndPoint.ToString());
server.LogDebug("Websocket initialized " + client.Socket.RemoteEndPoint.ToString());
foreach (var websocketModule in this.websocketModules)
{
websocketModule.ConnectionInitialized(server, this, client);
MTSC/ServerSide/Handlers/WebsocketRoutingHandler.cs
+1 -1
View File
@@ -185,7 +185,7 @@ namespace MTSC.ServerSide.Handlers
response.Headers[WebsocketHeaderAcceptKey] = returnBase64Key;
server.QueueMessage(client, response.GetPackedResponse(false));
client.Resources.SetResource(SocketState.Established);
server.LogDebug("Websocket initialized " + client.TcpClient.Client.RemoteEndPoint.ToString());
server.LogDebug("Websocket initialized " + client.Socket.RemoteEndPoint.ToString());
/*
* Create and assign route module to client.
*/
MTSC/ServerSide/Listeners/IListener.cs
+20
View File
@@ -0,0 +1,20 @@
using System.Net;
using System.Net.Sockets;
namespace MTSC.ServerSide.Listeners
{
/// <summary>
/// Interface for listeners.
/// </summary>
public interface IListener
{
bool Active { get; }
EndPoint LocalEndpoint { get; }
void Initialize(int port, IPAddress iPAddress);
Socket AcceptSocket();
bool Pending();
void Start();
void Stop();
}
}
MTSC/ServerSide/Listeners/TcpSocketListener.cs
+63
View File
@@ -0,0 +1,63 @@
using System;
using System.Collections.Concurrent;
using System.Net;
using System.Net.Sockets;
using System.Threading.Tasks;
namespace MTSC.ServerSide.Listeners
{
public sealed class TcpSocketListener : IListener
{
private readonly ConcurrentQueue<Socket> acceptedSockets = new();
private Socket socket;
public bool Active => this.socket is not null;
public EndPoint LocalEndpoint { get => this.socket?.LocalEndPoint; }
public void Initialize(int port, IPAddress ipAddress)
{
this.socket?.Close();
this.socket?.Dispose();
this.socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
this.socket.Bind(new IPEndPoint(ipAddress, port));
}
public Socket AcceptSocket()
{
if (this.acceptedSockets.TryDequeue(out var socket))
{
return socket;
}
throw new InvalidOperationException("Failed to accept client. No client available in queue");
}
public bool Pending()
{
return this.acceptedSockets.Count > 0;
}
public void Start()
{
this.socket.Listen(50);
var socket = this.socket;
Task.Run(async() =>
{
while(socket is not null)
{
var clientSocket = await this.socket.AcceptAsync();
this.acceptedSockets.Enqueue(clientSocket);
}
});
}
public void Stop()
{
this.socket?.Close();
this.socket?.Dispose();
this.socket = null;
}
}
}
MTSC/ServerSide/Server.cs
+30 -23
View File
@@ -2,6 +2,7 @@
using MTSC.Common;
using MTSC.Exceptions;
using MTSC.ServerSide.Handlers;
using MTSC.ServerSide.Listeners;
using MTSC.ServerSide.Schedulers;
using MTSC.ServerSide.UsageMonitors;
using Slim;
@@ -10,7 +11,6 @@ using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Text;
@@ -26,7 +26,6 @@ namespace MTSC.ServerSide
#region Fields
private bool running;
private X509Certificate2 certificate;
private TcpListener listener;
private readonly ProducerConsumerQueue<ClientData> addQueue = new();
private readonly List<ClientData> clients = new();
private readonly List<ClientData> toRemove = new();
@@ -42,6 +41,7 @@ namespace MTSC.ServerSide
private IConsumerQueue<(ClientData, byte[])> ConsumerMessageOutQueue { get => this.messageOutQueue; }
#endregion
#region Public Properties
public IListener Listener { get; set; } = new TcpSocketListener();
/// <summary>
/// Timeout for socket operations on clients
/// </summary>
@@ -85,7 +85,7 @@ namespace MTSC.ServerSide
/// <summary>
/// Returns the state of the server.
/// </summary>
public bool Running { get => this.listener != null; }
public bool Running => this.Listener?.Active is true;
/// <summary>
/// If set to true, requests client certificates.
/// </summary>
@@ -109,6 +109,8 @@ namespace MTSC.ServerSide
/// </summary>
public Server()
{
this.ServiceManager.RegisterServiceManager();
this.ServiceManager.RegisterSingleton<Server, Server>(sp => this);
}
/// <summary>
/// Creates an instance of server.
@@ -117,6 +119,8 @@ namespace MTSC.ServerSide
public Server(int port)
{
this.Port = port;
this.ServiceManager.RegisterServiceManager();
this.ServiceManager.RegisterSingleton<Server, Server>(sp => this);
}
/// <summary>
/// Creates an instance of server.
@@ -127,6 +131,8 @@ namespace MTSC.ServerSide
{
this.certificate = certificate;
this.Port = port;
this.ServiceManager.RegisterServiceManager();
this.ServiceManager.RegisterSingleton<Server, Server>(sp => this);
}
/// <summary>
/// Creates an instance of server.
@@ -137,6 +143,8 @@ namespace MTSC.ServerSide
{
this.IPAddress = ipAddress;
this.Port = port;
this.ServiceManager.RegisterServiceManager();
this.ServiceManager.RegisterSingleton<Server, Server>(sp => this);
}
/// <summary>
/// Creates an instance of server.
@@ -149,6 +157,8 @@ namespace MTSC.ServerSide
this.certificate = certificate;
this.Port = port;
this.IPAddress = ipAddress;
this.ServiceManager.RegisterServiceManager();
this.ServiceManager.RegisterSingleton<Server, Server>(sp => this);
}
#endregion
#region Public Methods
@@ -483,7 +493,7 @@ namespace MTSC.ServerSide
/// </summary>
public void Run()
{
this.listener?.Stop();
this.Listener?.Stop();
if (this.logger is null)
{
// Try to get a logger, in case it exists. If not, keep it null.
@@ -503,17 +513,15 @@ namespace MTSC.ServerSide
}
}
this.listener = new TcpListener(this.IPAddress, this.Port);
this.listener.Start();
this.Listener.Initialize(this.Port, this.IPAddress);
this.Listener.Start();
this.running = true;
this.Log("Server started on: " + this.listener.LocalEndpoint.ToString());
this.Log("Server started on: " + this.Listener.LocalEndpoint.ToString());
foreach(var toBeRunOnStartup in this.handlers.OfType<IRunOnStartup>())
{
toBeRunOnStartup.OnStartup(this);
}
this.ServiceManager.RegisterServiceManager();
this.ServiceManager.RegisterSingleton<Server, Server>(sp => this);
DateTime startLoopTime;
while (this.running)
{
@@ -540,10 +548,10 @@ namespace MTSC.ServerSide
*/
try
{
while (this.listener.Pending())
while (this.Listener.Pending())
{
var tcpClient = this.listener.AcceptTcpClient();
var clientStruct = new ClientData(tcpClient);
var client = this.Listener.AcceptSocket();
var clientStruct = new ClientData(client);
Task.Run(() => this.AcceptClient(clientStruct));
}
}
@@ -556,7 +564,7 @@ namespace MTSC.ServerSide
*/
while(this.ConsumerClientQueue.TryDequeue(out var client))
{
this.Log("Accepted new connection: " + client.TcpClient.Client.RemoteEndPoint.ToString());
this.Log("Accepted new connection: " + client.Socket.RemoteEndPoint.ToString());
this.clients.Add(client);
foreach (var handler in this.handlers)
{
@@ -607,13 +615,12 @@ namespace MTSC.ServerSide
}
}
this.listener.Stop();
this.Listener.Stop();
foreach (var client in this.Clients)
{
try
{
client?.SslStream?.Dispose();
client?.TcpClient?.Dispose();
client?.Dispose();
}
catch (Exception e)
{
@@ -624,7 +631,7 @@ namespace MTSC.ServerSide
}
}
this.listener = null;
this.Listener = null;
}
/// <summary>
/// Runs the server async.
@@ -660,9 +667,9 @@ namespace MTSC.ServerSide
handler.HandleSendMessage(this, client, ref sendMessage);
}
CommunicationPrimitives.SendMessage(client.TcpClient, sendMessage, client.SslStream);
CommunicationPrimitives.SendMessage(sendMessage, client.SafeNetworkStream, client.SslStream);
(client as IActiveClient).UpdateLastActivity();
this.LogDebug("Sent message to " + client.TcpClient.Client.RemoteEndPoint.ToString() +
this.LogDebug("Sent message to " + client.Socket.RemoteEndPoint.ToString() +
"\nMessage length: " + sendMessage.MessageLength);
}
catch(Exception e)
@@ -675,7 +682,7 @@ namespace MTSC.ServerSide
{
foreach (var client in this.Clients)
{
if (!client.TcpClient.Connected || client.ToBeRemoved)
if (!client.Socket.Connected || client.ToBeRemoved)
{
this.toRemove.Add(client);
}
@@ -690,7 +697,7 @@ namespace MTSC.ServerSide
handler.ClientRemoved(this, client);
}
this.LogDebug("Client removed: " + client.TcpClient?.Client?.RemoteEndPoint?.ToString());
this.LogDebug("Client removed: " + client.Socket?.RemoteEndPoint?.ToString());
client.Dispose();
}
catch(Exception e)
@@ -718,7 +725,7 @@ namespace MTSC.ServerSide
{
foreach(var client in this.Clients)
{
if (client.TcpClient.Available > 0 && !(client as IActiveClient).ReadingData)
if (client.Socket.Available > 0 && !(client as IActiveClient).ReadingData)
{
(client as IActiveClient).ReadingData = true;
Task.Run(async () =>
@@ -728,7 +735,7 @@ namespace MTSC.ServerSide
var timeout = this.ReadTimeout;
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.Socket.RemoteEndPoint as IPEndPoint} Message length: {message.MessageLength}");
if (this.LogMessageContents)
{
this.LogDebug(Encoding.UTF8.GetString(message.MessageBytes));