BSO-Mod/Shared/Connection.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Net.Sockets;
using System.Threading;

namespace BSO.Net
{
    /// <summary>Setzt aus einem Byte-Strom Nachrichten zusammen: [int Laenge][byte Typ][Payload].</summary>
    public class FrameParser
    {
        const int MaxFrame = 1024 * 1024;

        byte[] buffer = new byte[64 * 1024];
        int count;

        public void Append(byte[] data, int length)
        {
            EnsureCapacity(count + length);
            Buffer.BlockCopy(data, 0, buffer, count, length);
            count += length;
        }

        public void Parse(List<Message> messages)
        {
            int offset = 0;
            while (count - offset >= 5)
            {
                int length = BitConverter.ToInt32(buffer, offset);
                if (length < 1 || length > MaxFrame) throw new IOException("Invalid packet");
                if (count - offset - 4 < length) break;

                var payload = new byte[length - 1];
                Buffer.BlockCopy(buffer, offset + 5, payload, 0, payload.Length);
                messages.Add(new Message { Type = (MsgType)buffer[offset + 4], Payload = payload });
                offset += 4 + length;
            }
            if (offset > 0)
            {
                Buffer.BlockCopy(buffer, offset, buffer, 0, count - offset);
                count -= offset;
            }
        }

        public static byte[] Frame(MsgType type, byte[] payload)
        {
            payload ??= Array.Empty<byte>();
            var frame = new byte[5 + payload.Length];
            int length = payload.Length + 1; // Little Endian wie BitConverter
            frame[0] = (byte)length; frame[1] = (byte)(length >> 8); frame[2] = (byte)(length >> 16); frame[3] = (byte)(length >> 24);
            frame[4] = (byte)type;
            Buffer.BlockCopy(payload, 0, frame, 5, payload.Length);
            return frame;
        }

        void EnsureCapacity(int needed)
        {
            if (needed <= buffer.Length) return;
            var bigger = new byte[Math.Max(needed, buffer.Length * 2)];
            Buffer.BlockCopy(buffer, 0, bigger, 0, count);
            buffer = bigger;
        }
    }

    /// <summary>
    /// TCP-Verbindung wie bei MSCO: Lesen ist nicht-blockierend und wird ueber Poll() abgefragt (im Spiel im Hauptthread),
    /// Senden laeuft ueber eine Warteschlange und einen eigenen Thread - eine kurz volle Leitung haelt das Spiel nicht an.
    /// Kleine Pakete werden dabei zusammengefasst.
    /// </summary>
    public class Connection
    {
        volatile bool closed;
        volatile string closeReason;
        public bool IsClosed => closed;
        public string CloseReason => closeReason;

        // Mehr als das unterwegs = die Gegenseite nimmt nichts mehr an
        const long MaxQueuedBytes = 16L * 1024 * 1024;
        const int MaxBatchBytes = 256 * 1024;

        readonly TcpClient client;
        readonly NetworkStream stream;
        readonly FrameParser parser = new();
        readonly byte[] readBuffer = new byte[64 * 1024];
        readonly Queue<byte[]> outgoing = new();
        readonly object gate = new();
        long queuedBytes;
        readonly Thread writer;

        public Connection(TcpClient client)
        {
            this.client = client;
            client.NoDelay = true;
            stream = client.GetStream();
            writer = new Thread(WriteLoop) { IsBackground = true, Name = "BSO send" };
            writer.Start();
        }

        /// <summary>Sende-Thread: wartet auf Pakete und schreibt sie (gebuendelt) in die Verbindung.</summary>
        void WriteLoop()
        {
            var batch = new MemoryStream(MaxBatchBytes);
            try
            {
                while (true)
                {
                    batch.SetLength(0);
                    lock (gate)
                    {
                        while (outgoing.Count == 0 && !closed) Monitor.Wait(gate);
                        if (outgoing.Count == 0) return; // geschlossen und nichts mehr zu senden
                        do
                        {
                            byte[] frame = outgoing.Dequeue();
                            queuedBytes -= frame.Length;
                            batch.Write(frame, 0, frame.Length);
                        } while (outgoing.Count > 0 && batch.Length + outgoing.Peek().Length <= MaxBatchBytes);
                        Monitor.PulseAll(gate); // Close() wartet evtl. aufs Leerwerden
                    }
                    stream.Write(batch.GetBuffer(), 0, (int)batch.Length);
                }
            }
            catch (Exception e)
            {
                if (!closed) Close("Netzwerkfehler: " + e.Message);
            }
        }

        readonly List<Message> polled = new(); // wiederverwendet: der Aufrufer geht sie sofort durch
        public List<Message> Poll()
        {
            var messages = polled;
            messages.Clear();
            if (IsClosed) return messages;
            try
            {
                // Verbindung von der Gegenseite geschlossen?
                if (client.Client.Poll(0, SelectMode.SelectRead) && client.Available == 0)
                {
                    Close("Verbindung getrennt");
                    return messages;
                }
                while (stream.DataAvailable)
                {
                    int read = stream.Read(readBuffer, 0, readBuffer.Length);
                    if (read <= 0) break;
                    parser.Append(readBuffer, read);
                }
                parser.Parse(messages);
            }
            catch (Exception e)
            {
                Close("Netzwerkfehler: " + e.Message);
            }
            return messages;
        }

        public void Send(MsgType type, byte[] payload)
        {
            if (closed) return;
            byte[] frame = FrameParser.Frame(type, payload);
            bool tooMuch;
            lock (gate)
            {
                tooMuch = queuedBytes + frame.Length > MaxQueuedBytes;
                if (!tooMuch)
                {
                    outgoing.Enqueue(frame);
                    queuedBytes += frame.Length;
                    Monitor.Pulse(gate);
                }
            }
            if (tooMuch) Close("Verbindung zu langsam - zu viele Daten warten aufs Senden");
        }

        public void Close(string reason)
        {
            lock (gate)
            {
                if (closed) return;
                // Was schon in der Warteschlange liegt (z.B. eine Abschiedsnachricht), kurz noch rausgehen lassen
                var until = DateTime.UtcNow.AddMilliseconds(300);
                while (outgoing.Count > 0 && DateTime.UtcNow < until && Thread.CurrentThread != writer)
                    Monitor.Wait(gate, 50);
                closed = true;
                closeReason = reason;
                outgoing.Clear();
                queuedBytes = 0;
                Monitor.PulseAll(gate);
            }
            try { client.Close(); } catch { }
        }
    }
}