using System; using System.Collections.Generic; using System.IO; using System.Net.Sockets; using System.Threading; namespace BSO.Net { /// Setzt aus einem Byte-Strom Nachrichten zusammen: [int Laenge][byte Typ][Payload]. 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 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(); 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; } } /// /// 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. /// 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 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(); } /// Sende-Thread: wartet auf Pakete und schreibt sie (gebuendelt) in die Verbindung. 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 polled = new(); // wiederverwendet: der Aufrufer geht sie sofort durch public List 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 { } } } }