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
{
///
/// 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.
///
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 Buffer = new List();
public bool Closed;
}
readonly object _gate = new object();
readonly Queue _mainThread = new Queue();
TcpListener _listener;
TcpClient _client;
NetworkStream _clientStream;
readonly List _clientBuffer = new List();
readonly Dictionary _guests = new Dictionary();
int _nextPeer = 2;
Thread _acceptThread;
volatile bool _running;
/// Session bound to this transport (set before Host/Join).
public LuneSession Session;
/// True while a host listener or a client socket is open.
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
/// Delivers queued socket events on the game thread. Call once per frame.
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 buffer, Action 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();
}
}
}