Timerix/Meum
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ClientConnection changes

Browse Source
This commit is contained in:
Timerix 2025-07-05 01:21:18 +03:00
parent bf4924c4d6
commit 1b89a75ecb
6 changed files with 121 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
Meum.Client/ServerConnection.cs
View File

@ -1,21 +1,28 @@
using System.Net;
using System.Net.Quic;
using System.Runtime.InteropServices;
using DTLib.Logging;
using Meum.Core.Messages;
namespace Meum.Client;
public class ServerConnection : IAsyncDisposable
{
private readonly QuicConnection _quicConnection;
private readonly QuicStreamWrapper _systemStream;
private readonly ILogger _logger;
public DnsEndPoint ServerEndPoint { get; }
private ServerConnection(QuicConnection quicConnection, DnsEndPoint serverEndPoint, ILogger logger)
private ServerConnection(QuicConnection quicConnection,
DnsEndPoint serverEndPoint,
QuicStreamWrapper systemStream,
ILogger logger)
{
ServerEndPoint = serverEndPoint;
_quicConnection = quicConnection;
_systemStream = systemStream;
_logger = logger;
}
@ -24,13 +31,31 @@ public class ServerConnection : IAsyncDisposable
ILogger logger,
CancellationToken ct)
{
var serverConnection = new ServerConnection(quicConnection, serverEndPoint, logger);
var systemStream = await quicConnection.OpenStreamAsync(QuicStreamType.Bidirectional, ct);
var serverConnection = new ServerConnection(quicConnection, serverEndPoint, systemStream, logger);
await systemStream.SendPingReceivePong();
return serverConnection;
}
public async Task Register()
{
await _systemStream.WriteStructAsync(new DataMessageHeader(
MessageTypeCode.RegistrationRequest,
Marshal.SizeOf<RegistrationRequest>()));
var registrationRequest = new RegistrationRequest();
await _systemStream.WriteStructAsync(registrationRequest);
var responseHeader = await _systemStream.ReadDataMessageHeaderAsync();
if(responseHeader.type_code != MessageTypeCode.RegistrationResponse)
throw new Exception($"Invalid response header type: {responseHeader.type_code}");
}
public async Task Authorize()
{
}
public override int GetHashCode()
{
return _quicConnection.RemoteEndPoint.GetHashCode();

9
Meum.Core/Messages/RegistrationRequest.cs Normal file
View File

@ -0,0 +1,9 @@
using System.Runtime.InteropServices;
namespace Meum.Core.Messages;
[StructLayout(LayoutKind.Sequential)]
public record struct RegistrationRequest(byte[] hash)
{
}

4
Meum.Core/QuicStreamWrapper.cs
View File

@ -17,7 +17,7 @@ public class QuicStreamWrapper : IAsyncDisposable
public async ValueTask<T> ReadStructAsync<T>(CancellationToken ct = default)
where T : struct
{
byte[] buffer = ArrayPool<byte>.Shared.Rent(Marshal.SizeOf(typeof(T)));
byte[] buffer = ArrayPool<byte>.Shared.Rent(Marshal.SizeOf<T>());
var handle = GCHandle.Alloc(buffer, GCHandleType.Pinned);
try
{
@ -34,7 +34,7 @@ public class QuicStreamWrapper : IAsyncDisposable
public ValueTask WriteStructAsync<T>(T msg_struct, CancellationToken ct = default)
where T : struct
{
byte[] buffer = ArrayPool<byte>.Shared.Rent(Marshal.SizeOf(typeof(T)));
byte[] buffer = ArrayPool<byte>.Shared.Rent(Marshal.SizeOf<T>());
var handle = GCHandle.Alloc(buffer, GCHandleType.Pinned);
try
{

92
Meum.Server/ClientConnection.cs
View File

@ -8,47 +8,83 @@ public class ClientConnection : IAsyncDisposable
{
private readonly QuicConnection _quicConnection;
private QuicStreamWrapper _systemStream;
private ILogger _logger;
private ContextLogger _logger;
private CancellationTokenSource _connectionCts = new();
private ClientConnection(QuicConnection quicConnection, QuicStreamWrapper systemStream, ILogger logger)
public int Id { get; }
public bool IsAuthorized { get; private set; }
private ClientConnection(int id, QuicConnection quicConnection, QuicStreamWrapper systemStream, ILogger logger)
{
Id = id;
_quicConnection = quicConnection;
_systemStream = systemStream;
_logger = logger;
_logger = new ContextLogger($"Connection-{id}", logger);
}
public static async Task<ClientConnection> OpenAsync(
QuicConnection quicConnection,
ILogger logger,
CancellationToken ct)
public static async Task<ClientConnection> OpenAsync(int id,
QuicConnection quicConnection, ILogger logger, CancellationToken ct)
{
var systemStream = await quicConnection.AcceptStreamAsync(QuicStreamType.Bidirectional, ct);
await systemStream.ReceivePingSendPong();
var clientConnection = new ClientConnection(quicConnection, systemStream, logger);
DataMessageHeader header = await systemStream.ReadDataMessageHeaderAsync(ct);
switch (header.type_code)
{
case MessageTypeCode.RegistrationRequest:
break;
case MessageTypeCode.AuthorizationRequest:
// if (authorized)
// clientConnection.HandleClientRequestsAsync()
break;
default:
throw new Exception($"New connection sent unexpected message: {header.type_code}");
}
var clientConnection = new ClientConnection(id, quicConnection, systemStream, logger);
clientConnection.HandleRequestsAsync(ct);
return clientConnection;
}
private async void HandleRequestsAsync(CancellationToken ct)
{
try
{
while (!_connectionCts.IsCancellationRequested)
{
try
{
var header = await _systemStream.ReadDataMessageHeaderAsync(ct);
if(IsAuthorized)
HandleAuthorizedRequest(header);
else HandleUnauthorizedRequest(header);
}
catch (Exception ex)
{
_logger.LogWarn(ex);
}
}
// private async void HandleClientRequestsAsync(CancellationToken ct = default)
// {
//
// }
await DisposeAsync();
}
catch (Exception ex)
{
_logger.LogError(ex);
}
}
private void HandleUnauthorizedRequest(DataMessageHeader header)
{
switch (header.type_code)
{
default:
throw new Exception($"New connection sent unexpected message: {header.type_code}");
case MessageTypeCode.RegistrationRequest:
// TODO: registration
break;
case MessageTypeCode.AuthorizationRequest:
// TODO: authorization
IsAuthorized = true;
break;
}
}
private void HandleAuthorizedRequest(DataMessageHeader header)
{
switch (header.type_code)
{
default:
throw new Exception($"New connection sent unexpected message: {header.type_code}");
}
}
public override int GetHashCode() => Id;
public async ValueTask DisposeAsync()
{

7
Meum.Server/Program.cs
View File

@ -19,7 +19,9 @@ static class Program
try
{
var config = ServerConfig.LoadOrCreate(config_path);
var logger = new ConsoleLogger();
var logger = new CompositeLogger(
new ConsoleLogger(),
new FileLogger("logs", "meum-server"));
Functions.InitMsQuic(logger);
var server = new Server(config, logger);
@ -29,11 +31,10 @@ static class Program
try
{
var conn = await server.AcceptConnectionAsync();
}
catch (Exception ex)
{
logger.LogError("Main", ex.ToStringDemystified());
logger.LogError("MainLoop", ex);
}
}
}

25
Meum.Server/Server.cs
View File

@ -12,7 +12,7 @@ public class Server
private readonly X509Certificate _certificate;
private readonly ILogger _logger;
private QuicListener? _listener;
private int _lastConnectionId;
public Server(ServerConfig config, ILogger logger, X509Certificate? certificate = null)
{
@ -50,17 +50,16 @@ public class Server
if (_listener == null)
throw new Exception("Server is not listening");
while (true)
{
ct.ThrowIfCancellationRequested();
var quicConnection = await _listener.AcceptConnectionAsync(ct);
var timeOutCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
timeOutCts.CancelAfter(Constants.ConnectionTimeout);
var clientConnection = await ClientConnection.OpenAsync(
quicConnection,
_logger,
timeOutCts.Token);
return clientConnection;
}
ct.ThrowIfCancellationRequested();
var quicConnection = await _listener.AcceptConnectionAsync(ct);
var timeOutCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
timeOutCts.CancelAfter(Constants.ConnectionTimeout);
var clientConnection = await ClientConnection.OpenAsync(
_lastConnectionId,
quicConnection,
_logger,
timeOutCts.Token);
Interlocked.Increment(ref _lastConnectionId);
return clientConnection;
}
}