Convert TCP networking to sync and binary, add DHCP and IP stack init on boot

2026-09-05 23:41:28 +02:00 · 2026-06-18 23:54:23 +02:00
parent 79bc87f5c7
commit ba5d670d1a
8 changed files with 345 additions and 143 deletions
@@ -7,12 +7,11 @@ public class TcpConnection(TcpClient client) : IDisposable
 {
    public TcpClient Client { get; } = client;
    public NetworkStream Stream { get; } = client.GetStream();
-    public SemaphoreSlim SendLock { get; } = new(1, 1);
+    public object SendLock { get; } = new();

    public void Dispose()
    {
        Stream.Dispose();
        Client.Dispose();
-        SendLock.Dispose();
    }
 }
@@ -2,9 +2,8 @@ using RemSox.Networking;

 using System.Buffers.Binary;
 using System.Collections.Concurrent;
-using System.Diagnostics.CodeAnalysis;
 using System.Net.Sockets;
-using System.Text.Json;
+using System.Text;

 public abstract class TcpRpcBase(IPacketCrypto? crypto = null)
 {
@@ -15,57 +14,65 @@ public abstract class TcpRpcBase(IPacketCrypto? crypto = null)
    // Security Guard: Prevent OOM/DoS via oversized length headers (Default: 32MB)
    protected const int MaxMessageSize = 32 * 1024 * 1024;

-    /// <summary>
-    /// Base message registration for handlers tracking the sending connection.
-    /// </summary>
-    [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Handled via library constraints")]
-    public void ListenTo<T>(string type, Func<TcpConnection, T, Task> handler)
+    public void ListenTo(string type, Func<TcpConnection, byte[], Task> handler)
+    {
+        handlers[type] = async (conn, msg) => await handler(conn, msg.Payload);
+    }
+
+    public void ListenTo(string type, Func<byte[], Task> handler)
+    {
+        handlers[type] = async (conn, msg) => await handler(msg.Payload);
+    }
+
+    public void RespondTo(string type, Func<byte[], Task<byte[]>> handler)
    {
        handlers[type] = async (conn, msg) =>
        {
-            T data = JsonSerializer.Deserialize<T>(msg.Payload)!;
-            await handler(conn, data);
+            byte[] res = await handler(msg.Payload);
+            SendRaw(conn, new TcpMessage
+            {
+                Type = type,
+                RequestId = msg.RequestId,
+                Payload = res
+            });
        };
    }

-    /// <summary>
-    /// Overloaded listener where connection mapping can be ignored (Convenient for Clients).
-    /// </summary>
-    public void ListenTo<T>(string type, Func<T, Task> handler)
+    protected void SendRaw(TcpConnection conn, TcpMessage msg)
    {
-        ListenTo<T>(type, async (_, data) => await handler(data));
-    }
-
-    [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Handled via library constraints")]
-    protected async Task SendRaw(TcpConnection conn, TcpMessage msg)
-    {
-        byte[] data = JsonSerializer.SerializeToUtf8Bytes(msg);
+        byte[] rawData = SerializeMessage(msg);

        if (crypto is not null)
        {
-            data = crypto.Encrypt(data);
+            rawData = crypto.Encrypt(rawData);
        }

-        // Allocate a single contiguous frame buffer to prevent inter-thread fragmentation 
-        // and eliminate redundant Socket Write Syscalls.
-        byte[] packet = new byte[4 + data.Length];
-        BinaryPrimitives.WriteInt32LittleEndian(packet, data.Length);
-        Array.Copy(data, 0, packet, 4, data.Length);
+        byte[] packet = new byte[4 + rawData.Length];
+        BinaryPrimitives.WriteInt32LittleEndian(packet, rawData.Length);
+        Array.Copy(rawData, 0, packet, 4, rawData.Length);

-        // Enforce sequence safety across multiple threads pushing data out of a singular stream
-        await conn.SendLock.WaitAsync();
-        try
+        lock (conn.SendLock)
        {
-            await conn.Stream.WriteAsync(packet);
-        }
-        finally
-        {
-            _ = conn.SendLock.Release();
+            conn.Stream.Write(packet, 0, packet.Length);
        }
    }

-    [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Handled via library constraints")]
-    protected async Task HandleConnection(TcpConnection conn, CancellationToken token = default)
+    private static void ReadExact(NetworkStream stream, byte[] buffer, int offset, int count)
+    {
+        int totalRead = 0;
+        while (totalRead < count)
+        {
+            int read = stream.Read(buffer, offset + totalRead, count - totalRead);
+            if (read == 0)
+            {
+                Thread.Sleep(10);
+                continue;
+            }
+            totalRead += read;
+        }
+    }
+
+    protected void HandleConnection(TcpConnection conn, CancellationToken token = default)
    {
        NetworkStream stream = conn.Stream;
        byte[] lengthBytes = new byte[4];
@@ -74,32 +81,41 @@ public abstract class TcpRpcBase(IPacketCrypto? crypto = null)
        {
            while (!token.IsCancellationRequested && conn.Client.Connected)
            {
-                // 1. Frame Length Read
-                await stream.ReadExactlyAsync(lengthBytes, token);
+                Console.WriteLine("[TcpRpc] Waiting for incoming packet...");
+
+                ReadExact(stream, lengthBytes, 0, 4);
                int length = BinaryPrimitives.ReadInt32LittleEndian(lengthBytes);

-                // DoS Payload Protection Check
                if (length is <= 0 or > MaxMessageSize)
                {
-                    throw new InvalidDataException($"Protocol violation: Received packet length of {length} bytes exceeds limits.");
+                    Console.WriteLine($"[TcpRpc] Invalid packet length: {length}");
+                    Thread.Sleep(10);
+                    continue;
                }

-                // 2. Body Payload Read
+                Console.WriteLine($"[TcpRpc] Incoming packet length: {length} bytes");
+
                byte[] buffer = new byte[length];
-                await stream.ReadExactlyAsync(buffer, token);
+                ReadExact(stream, buffer, 0, length);
+
+                Console.WriteLine($"[TcpRpc] Incoming packet payload: {BitConverter.ToString(buffer)}");

                if (crypto is not null)
                {
                    buffer = crypto.Decrypt(buffer);
                }

-                TcpMessage? msg = JsonSerializer.Deserialize<TcpMessage>(buffer);
+                Console.WriteLine($"[TcpRpc] Decrypted packet payload: {BitConverter.ToString(buffer)}");
+
+                TcpMessage? msg = DeserializeMessage(buffer);
                if (msg is null)
                {
+                    Console.WriteLine("[TcpRpc] Failed to deserialize incoming message.");
                    continue;
                }

-                // 3. Response Handler Check
+                Console.WriteLine($"[TcpRpc] Incoming message type: {msg.Type}, requestId: {msg.RequestId}, payload length: {msg.Payload.Length} bytes");
+
                if (pendingRequests.TryGetValue(msg.RequestId, out TaskCompletionSource<byte[]>? tcs))
                {
                    _ = tcs.TrySetResult(msg.Payload);
@@ -107,11 +123,8 @@ public abstract class TcpRpcBase(IPacketCrypto? crypto = null)
                    continue;
                }

-                // 4. Inbound Router Handler
                if (handlers.TryGetValue(msg.Type, out Func<TcpConnection, TcpMessage, Task>? handler))
                {
-                    // Decouple handler execution to the ThreadPool so slow business logic 
-                    // doesn't bottleneck packet processing from the network stream interface.
                    _ = Task.Run(async () =>
                    {
                        try
@@ -120,7 +133,6 @@ public abstract class TcpRpcBase(IPacketCrypto? crypto = null)
                        }
                        catch
                        {
-                            // Operational log placement here for user exceptions inside delegates
                        }
                    }, token);
                }
@@ -128,7 +140,6 @@ public abstract class TcpRpcBase(IPacketCrypto? crypto = null)
        }
        catch
        {
-            // Explicitly handles natural dropping scenarios cleanly
        }
        finally
        {
@@ -137,5 +148,90 @@ public abstract class TcpRpcBase(IPacketCrypto? crypto = null)
        }
    }

+    private static byte[] SerializeMessage(TcpMessage msg)
+    {
+        byte[] typeBytes = Encoding.UTF8.GetBytes(msg.Type);
+        byte[] requestIdBytes = Encoding.UTF8.GetBytes(msg.RequestId);
+
+        byte[] data = new byte[4 + typeBytes.Length + 4 + requestIdBytes.Length + 4 + msg.Payload.Length];
+        int offset = 0;
+
+        WriteInt32(data, ref offset, typeBytes.Length);
+        typeBytes.CopyTo(data, offset);
+        offset += typeBytes.Length;
+
+        WriteInt32(data, ref offset, requestIdBytes.Length);
+        requestIdBytes.CopyTo(data, offset);
+        offset += requestIdBytes.Length;
+
+        WriteInt32(data, ref offset, msg.Payload.Length);
+        msg.Payload.CopyTo(data, offset);
+
+        return data;
+    }
+
+    private static TcpMessage? DeserializeMessage(byte[] data)
+    {
+        int offset = 0;
+        if (offset + 4 > data.Length)
+        {
+            return null;
+        }
+
+        int typeLen = ReadInt32(data, ref offset);
+        if (offset + typeLen > data.Length)
+        {
+            return null;
+        }
+
+        string type = Encoding.UTF8.GetString(data, offset, typeLen);
+        offset += typeLen;
+
+        if (offset + 4 > data.Length)
+        {
+            return null;
+        }
+
+        int requestIdLen = ReadInt32(data, ref offset);
+        if (offset + requestIdLen > data.Length)
+        {
+            return null;
+        }
+
+        string requestId = Encoding.UTF8.GetString(data, offset, requestIdLen);
+        offset += requestIdLen;
+
+        if (offset + 4 > data.Length)
+        {
+            return null;
+        }
+
+        int payloadLen = ReadInt32(data, ref offset);
+        if (offset + payloadLen > data.Length)
+        {
+            return null;
+        }
+
+        byte[] payload = new byte[payloadLen];
+        Array.Copy(data, offset, payload, 0, payloadLen);
+
+        return new TcpMessage { Type = type, RequestId = requestId, Payload = payload };
+    }
+
+    private static void WriteInt32(byte[] data, ref int offset, int value)
+    {
+        data[offset++] = (byte)(value & 0xFF);
+        data[offset++] = (byte)((value >> 8) & 0xFF);
+        data[offset++] = (byte)((value >> 16) & 0xFF);
+        data[offset++] = (byte)((value >> 24) & 0xFF);
+    }
+
+    private static int ReadInt32(byte[] data, ref int offset)
+    {
+        int val = data[offset] | (data[offset + 1] << 8) | (data[offset + 2] << 16) | (data[offset + 3] << 24);
+        offset += 4;
+        return val;
+    }
+
    protected abstract void OnConnectionClosed(TcpConnection conn);
 }
@@ -1,6 +1,5 @@
-using System.Diagnostics.CodeAnalysis;
+using System.Net;
 using System.Net.Sockets;
-using System.Text.Json;

 namespace RemSox.Networking;

@@ -8,19 +7,17 @@ public class TcpRpcClient(IPacketCrypto? crypto = null) : TcpRpcBase(crypto)
 {
    private TcpConnection? connection;

-    public async Task ConnectAsync(string host, int port)
+    public void Connect(IPAddress ipAddress, int port)
    {
        TcpClient client = new();
-        await client.ConnectAsync(host, port);
+        client.Connect(ipAddress, port);

        connection = new TcpConnection(client);

-        // Run network monitoring task loop background-detached
-        _ = HandleConnection(connection);
+        _ = Task.Run(() => HandleConnection(connection));
    }

-    [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Handled via library constraints")]
-    public async Task<TRes> RequestAsync<TReq, TRes>(string type, TReq request, TimeSpan timeout = default)
+    public async Task<byte[]> RequestAsync(string type, byte[] request, TimeSpan timeout = default)
    {
        if (connection is null)
        {
@@ -29,7 +26,7 @@ public class TcpRpcClient(IPacketCrypto? crypto = null) : TcpRpcBase(crypto)

        if (timeout == default)
        {
-            timeout = TimeSpan.FromSeconds(30); // Default fallback timeout
+            timeout = TimeSpan.FromSeconds(30);
        }

        string requestId = Guid.NewGuid().ToString();
@@ -38,19 +35,17 @@ public class TcpRpcClient(IPacketCrypto? crypto = null) : TcpRpcBase(crypto)

        try
        {
-            await SendRaw(connection, new TcpMessage
+            SendRaw(connection, new TcpMessage
            {
                Type = type,
                RequestId = requestId,
-                Payload = JsonSerializer.SerializeToUtf8Bytes(request)
+                Payload = request
            });

-            // Enforce async timeout safety to prevent permanent dictionary leaks on dropped calls
            using CancellationTokenSource timeoutCts = new(timeout);
-            await using (timeoutCts.Token.Register(() => tcs.TrySetCanceled()))
+            using (timeoutCts.Token.Register(() => tcs.TrySetCanceled()))
            {
-                byte[] responseBytes = await tcs.Task;
-                return JsonSerializer.Deserialize<TRes>(responseBytes)!;
+                return await tcs.Task;
            }
        }
        finally
@@ -59,19 +54,18 @@ public class TcpRpcClient(IPacketCrypto? crypto = null) : TcpRpcBase(crypto)
        }
    }

-    [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Handled via library constraints")]
-    public Task SendAsync<T>(string type, T data)
+    public void Send(string type, byte[] data)
    {
        if (connection is null)
        {
            throw new InvalidOperationException("Client not connected.");
        }

-        return SendRaw(connection, new TcpMessage
+        SendRaw(connection, new TcpMessage
        {
            Type = type,
            RequestId = Guid.NewGuid().ToString(),
-            Payload = JsonSerializer.SerializeToUtf8Bytes(data)
+            Payload = data
        });
    }

@@ -82,7 +76,6 @@ public class TcpRpcClient(IPacketCrypto? crypto = null) : TcpRpcBase(crypto)
            connection = null;
        }

-        // Fail-fast all lingering tasks waiting on an dead connection loop
        foreach (TaskCompletionSource<byte[]> req in pendingRequests.Values)
        {
            _ = req.TrySetException(new SocketException((int)SocketError.ConnectionReset));
@@ -1,19 +1,17 @@
 using System.Collections.Concurrent;
-using System.Diagnostics.CodeAnalysis;
 using System.Net;
 using System.Net.Sockets;
-using System.Text.Json;

 namespace RemSox.Networking;

 public class TcpRpcServer(IPacketCrypto? crypto = null) : TcpRpcBase(crypto)
 {
-    // Fix: Using ConcurrentDictionary to prevent collection errors during structural additions/prunings
    private readonly ConcurrentDictionary<TcpConnection, byte> connections = new();
+    private readonly ConcurrentDictionary<string, byte> activeEndpoints = new();
    private TcpListener? listener;
    private CancellationTokenSource? cts;

-    public async Task StartAsync(int port, CancellationToken token = default)
+    public void StartAsync(int port, CancellationToken token = default)
    {
        cts = CancellationTokenSource.CreateLinkedTokenSource(token);

@@ -24,17 +22,25 @@ public class TcpRpcServer(IPacketCrypto? crypto = null) : TcpRpcBase(crypto)
        {
            while (!cts.IsCancellationRequested)
            {
-                TcpClient client = await listener.AcceptTcpClientAsync(cts.Token);
+                TcpClient client = listener.AcceptTcpClient();
+                string ep = client.Client.RemoteEndPoint?.ToString() ?? "";

+                if (!activeEndpoints.TryAdd(ep, 0))
+                {
+                    client.Close();
+                    Thread.Sleep(10);
+                    continue;
+                }
+
+                Console.WriteLine($"Client connected: {ep}");
                TcpConnection conn = new(client);
                _ = connections.TryAdd(conn, 0);

-                _ = HandleConnection(conn, cts.Token);
+                _ = Task.Run(() => HandleConnection(conn, cts.Token));
            }
        }
        catch (OperationCanceledException)
        {
-            // Expected shutdown scenario
        }
    }

@@ -51,62 +57,26 @@ public class TcpRpcServer(IPacketCrypto? crypto = null) : TcpRpcBase(crypto)
        connections.Clear();
    }

-    [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Handled via library constraints")]
-    public void RespondTo<TReq, TRes>(string type, Func<TReq, Task<TRes>> handler)
+    public void SendRawToAll(string type, byte[] payload)
    {
-        handlers[type] = async (conn, msg) =>
-        {
-            TReq req = JsonSerializer.Deserialize<TReq>(msg.Payload)!;
-            TRes res = await handler(req);
-
-            await SendRaw(conn, new TcpMessage
-            {
-                Type = type,
-                RequestId = msg.RequestId,
-                Payload = JsonSerializer.SerializeToUtf8Bytes(res)
-            });
-        };
-    }
-
-    [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Handled via library constraints")]
-    public Task SendToAll<T>(string type, T data)
-    {
-        byte[] payloadBytes = JsonSerializer.SerializeToUtf8Bytes(data);
-        List<Task> sendTasks = [];
-
        foreach (TcpConnection conn in connections.Keys)
        {
-            sendTasks.Add(SendRaw(conn, new TcpMessage
-            {
-                Type = type,
-                RequestId = Guid.NewGuid().ToString(),
-                Payload = payloadBytes
-            }));
-        }
-
-        return Task.WhenAll(sendTasks);
-    }
-
-    /// <summary> Broadcasts raw binary payload to all connected clients. </summary>
-    public Task SendRawToAll(string type, byte[] payload)
-    {
-        List<Task> sendTasks = [];
-
-        foreach (TcpConnection conn in connections.Keys)
-        {
-            sendTasks.Add(SendRaw(conn, new TcpMessage
+            SendRaw(conn, new TcpMessage
            {
                Type = type,
                RequestId = Guid.NewGuid().ToString(),
                Payload = payload
-            }));
+            });
        }
-
-        return Task.WhenAll(sendTasks);
    }

    protected override void OnConnectionClosed(TcpConnection conn)
    {
        _ = connections.TryRemove(conn, out _);
+
+        string ep = conn.Client.Client.RemoteEndPoint?.ToString() ?? "";
+        _ = activeEndpoints.TryRemove(ep, out _);
+
+        Console.WriteLine($"Client disconnected: {ep}");
    }
 }