PeerConnectiveChat/PCC.Shared/App/Networking/TcpPeer.cs

using System.Buffers;
using Microsoft.AspNetCore.Connections;
using PCC.App.BinaryFormatter;

namespace PCC.App.Networking;

public class TcpPeer(ConnectionContext context)
{
    public async Task SendBlockAsync(ReadOnlyMemory<byte> block)
    {
        context.Transport.Output.WriteBlock(block);
        await context.Transport.Output.FlushAsync();
    }

    public async Task SendBlockAsync(byte[] block)
    {
        context.Transport.Output.WriteBlock(block);
        await context.Transport.Output.FlushAsync();
    }

    public async Task<byte[]?> RxBlockAsync(CancellationToken? timeOut = null)
    {
        while (true)
        {
            var result = await context.Transport.Input.ReadAsync(timeOut ?? CancellationToken.None);
            if (result.IsCanceled) break;
            var block = ReadBlock(result.Buffer);
            if (block != null) return block;
            if (result.IsCompleted) break;
        }

        return null;
    }

    private byte[]? ReadBlock(ReadOnlySequence<byte> rx)
    {
        var reader = new SequenceReader<byte>(rx);

        var lenR = reader.TryRead7BitEncodedInt(out var len);

        if (lenR.HasValue == false)
        {
            context.Transport.Input.AdvanceTo(rx.Start, reader.Position);
            return null;
        }

        if (lenR == false) throw new InvalidDataException("Invalid 7Bit Encoded Int");

        var requestLen = (int)reader.Consumed + len;
        if (rx.Length < requestLen)
        {
            context.Transport.Input.AdvanceTo(rx.Start, rx.End);
            return null;
        }

        reader.TryReadExact(len, out var seq);

        var block = seq.ToArray();

        context.Transport.Input.AdvanceTo(reader.Position);

        return block;
    }

    public void Disconnect()
    {
        context.Transport.Input.Complete();
        context.Transport.Output.Complete();
    }
}