champions-of-luneraria/Assets/Luneraria/Net/LuneTcp.cs

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

namespace Luneraria.Net
{
    /// <summary>
    /// TCP transport for LuneSession (the ENet peer of the reference). Reliable messages are
    /// length-prefixed UTF-8 lines (LuneWire); combat frames are binary and tagged so a late
    /// frame can be dropped by the session's serial. One socket per guest, peer 1 is the host.
    /// Pump() must be called from the game loop; sockets never call into the session on their own.
    /// </summary>
    public sealed class LuneTcp : ILuneNetLink, IDisposable
    {
        const byte TagMessage = 1;
        const byte TagFrame = 2;

        sealed class Guest
        {
            public int Peer;
            public TcpClient Client;
            public NetworkStream Stream;
            public readonly List<byte> Buffer = new List<byte>();
            public bool Closed;
        }

        readonly object _gate = new object();
        readonly Queue<Action> _mainThread = new Queue<Action>();
        TcpListener _listener;
        TcpClient _client;
        NetworkStream _clientStream;
        readonly List<byte> _clientBuffer = new List<byte>();
        readonly Dictionary<int, Guest> _guests = new Dictionary<int, Guest>();
        int _nextPeer = 2;
        Thread _acceptThread;
        volatile bool _running;

        /// <summary>Session bound to this transport (set before Host/Join).</summary>
        public LuneSession Session;
        /// <summary>True while a host listener or a client socket is open.</summary>
        public bool Open { get { return _running; } }
        public string LastError = "";

        // ---------------------------------------------------------------- host
        public bool Host(int port)
        {
            try
            {
                _listener = new TcpListener(IPAddress.Any, port);
                _listener.Start();
                _running = true;
                _acceptThread = new Thread(AceptLoop) { IsBackground = true, Name = "LuneTcp accept" };
                _acceptThread.Start();
                return true;
            }
            catch (Exception error)
            {
                LastError = "Server auf Port " + port + " nicht startbar (" + error.Message + ").";
                return false;
            }
        }

        void AceptLoop()
        {
            while (_running && _listener != null)
            {
                TcpClient client;
                try
                {
                    client = _listener.AcceptTcpClient();
                }
                catch
                {
                    break;
                }
                lock (_gate)
                {
                    int peer = _nextPeer++;
                    var guest = new Guest { Peer = peer, Client = client, Stream = client.GetStream() };
                    _guests[peer] = guest;
                    _mainThread.Enqueue(() =>
                    {
                        if (Session != null)
                            Session.OnMessage(peer, "__connected", null);
                    });
                    var reader = new Thread(() => ReadLoop(guest)) { IsBackground = true, Name = "LuneTcp peer " + peer };
                    reader.Start();
                }
            }
        }

        // ---------------------------------------------------------------- client
        public bool Join(string address, int port)
        {
            try
            {
                _client = new TcpClient();
                _client.Connect(address, port);
                _clientStream = _client.GetStream();
                _running = true;
                var reader = new Thread(ClientReadLoop) { IsBackground = true, Name = "LuneTcp client" };
                reader.Start();
                _mainThread.Enqueue(() =>
                {
                    if (Session != null)
                        Session.OnClientConnected();
                });
                return true;
            }
            catch (Exception error)
            {
                LastError = "Verbindung zu " + address + ":" + port + " nicht möglich (" + error.Message + ").";
                _running = false;
                return false;
            }
        }

        // ---------------------------------------------------------------- pump
        /// <summary>Delivers queued socket events on the game thread. Call once per frame.</summary>
        public void Pump()
        {
            while (true)
            {
                Action action;
                lock (_gate)
                {
                    if (_mainThread.Count == 0)
                        break;
                    action = _mainThread.Dequeue();
                }
                action();
            }
        }

        // ---------------------------------------------------------------- ILuneNetLink
        public void Send(int peer, string kind, object payload)
        {
            byte[] body = System.Text.Encoding.UTF8.GetBytes(LuneWire.Encode(kind, payload));
            byte[] packet = Frame(TagMessage, body);
            if (peer == 1)
            {
                WriteClient(packet);
                return;
            }
            Guest guest;
            lock (_gate)
            {
                if (!_guests.TryGetValue(peer, out guest) || guest.Closed)
                    return;
            }
            try
            {
                guest.Stream.Write(packet, 0, packet.Length);
            }
            catch
            {
                QueueDrop(guest.Peer);
            }
        }

        public void SendFrame(int peer, int serial, double elapsed, float[] units, float[] buildingHp)
        {
            byte[] body = LuneWire.EncodeFrame(serial, elapsed, units, buildingHp);
            byte[] packet = Frame(TagFrame, body);
            if (peer == 1)
            {
                WriteClient(packet);
                return;
            }
            Guest guest;
            lock (_gate)
            {
                if (!_guests.TryGetValue(peer, out guest) || guest.Closed)
                    return;
            }
            try
            {
                guest.Stream.Write(packet, 0, packet.Length);
            }
            catch
            {
                QueueDrop(guest.Peer);
            }
        }

        void WriteClient(byte[] packet)
        {
            try
            {
                if (_clientStream != null)
                    _clientStream.Write(packet, 0, packet.Length);
            }
            catch
            {
                _mainThread.Enqueue(() =>
                {
                    if (Session != null)
                        Session.OnServerLost();
                });
            }
        }

        static byte[] Frame(byte tag, byte[] body)
        {
            var packet = new byte[5 + body.Length];
            packet[0] = tag;
            packet[1] = (byte)(body.Length & 0xff);
            packet[2] = (byte)((body.Length >> 8) & 0xff);
            packet[3] = (byte)((body.Length >> 16) & 0xff);
            packet[4] = (byte)((body.Length >> 24) & 0xff);
            Buffer.BlockCopy(body, 0, packet, 5, body.Length);
            return packet;
        }

        void ReadLoop(Guest guest)
        {
            var chunk = new byte[8192];
            while (_running && !guest.Closed)
            {
                int read;
                try
                {
                    read = guest.Stream.Read(chunk, 0, chunk.Length);
                }
                catch
                {
                    read = 0;
                }
                if (read <= 0)
                    break;
                for (int i = 0; i < read; i++)
                    guest.Buffer.Add(chunk[i]);
                Drain(guest.Buffer, (tag, body) => Deliver(guest.Peer, tag, body));
            }
            QueueDrop(guest.Peer);
        }

        void ClientReadLoop()
        {
            var chunk = new byte[8192];
            while (_running)
            {
                int read;
                try
                {
                    read = _clientStream.Read(chunk, 0, chunk.Length);
                }
                catch
                {
                    read = 0;
                }
                if (read <= 0)
                    break;
                for (int i = 0; i < read; i++)
                    _clientBuffer.Add(chunk[i]);
                Drain(_clientBuffer, (tag, body) => Deliver(1, tag, body));
            }
            _mainThread.Enqueue(() =>
            {
                if (Session != null)
                    Session.OnServerLost();
            });
        }

        static void Drain(List<byte> buffer, Action<byte, byte[]> emit)
        {
            while (buffer.Count >= 5)
            {
                int length = buffer[1] | (buffer[2] << 8) | (buffer[3] << 16) | (buffer[4] << 24);
                if (length < 0 || buffer.Count < 5 + length)
                    break;
                byte tag = buffer[0];
                var body = new byte[length];
                for (int i = 0; i < length; i++)
                    body[i] = buffer[5 + i];
                buffer.RemoveRange(0, 5 + length);
                emit(tag, body);
            }
        }

        void Deliver(int peer, byte tag, byte[] body)
        {
            if (tag == TagFrame)
            {
                int serial;
                double elapsed;
                float[] units;
                float[] buildingHp;
                LuneWire.DecodeFrame(body, out serial, out elapsed, out units, out buildingHp);
                _mainThread.Enqueue(() =>
                {
                    if (Session != null)
                        Session.OnFrame(peer, serial, elapsed, units, buildingHp);
                });
                return;
            }
            string line = System.Text.Encoding.UTF8.GetString(body);
            _mainThread.Enqueue(() =>
            {
                if (Session == null)
                    return;
                string kind;
                object payload;
                LuneWire.Decode(line, out kind, out payload);
                Session.OnMessage(peer, kind, payload);
            });
        }

        void QueueDrop(int peer)
        {
            _mainThread.Enqueue(() =>
            {
                Guest guest;
                lock (_gate)
                {
                    if (_guests.TryGetValue(peer, out guest))
                    {
                        guest.Closed = true;
                        _guests.Remove(peer);
                    }
                }
                if (Session != null)
                    Session.OnPeerDisconnected(peer);
            });
        }

        public void Close()
        {
            _running = false;
            try
            {
                if (_listener != null)
                    _listener.Stop();
            }
            catch
            {
            }
            _listener = null;
            try
            {
                if (_client != null)
                    _client.Close();
            }
            catch
            {
            }
            _client = null;
            _clientStream = null;
            lock (_gate)
            {
                foreach (Guest guest in _guests.Values)
                {
                    guest.Closed = true;
                    try
                    {
                        guest.Client.Close();
                    }
                    catch
                    {
                    }
                }
                _guests.Clear();
            }
        }

        public void Dispose()
        {
            Close();
        }
    }
}