using UnityEngine; using System.Collections.Generic; using System.Net; using System.Threading; using System; using NeonTea.Quakeball.TeaNet.Packets; namespace NeonTea.Quakeball.TeaNet.Peers { /// Manages connections for Peer, sends them keepalives and sends and handles incoming messages. public class ConnectionManager { private ulong ConnectionCounter; private Dictionary Connections = new Dictionary(); private Dictionary IPtoID = new Dictionary(); private Dictionary> PacketQueue = new Dictionary>(); private Peer Peer; public Dictionary> ProtocolActionQueues = new Dictionary>(); private Thread UpdateThread; public long Timeout = 8000; public long Interval = 100; public ConnectionManager(Peer peer) { Peer = peer; UpdateThread = new Thread(new ThreadStart(UpdateThreadMethod)); UpdateThread.Start(); } public void StopThread() { UpdateThread.Abort(); } /// Find a given Connection. Should not be used, unless expecting to establish a connection with the endpoint. public Connection Find(IPEndPoint endpoint) { if (IPtoID.ContainsKey(endpoint)) { return Connections[IPtoID[endpoint]]; } Connection conn = new Connection(endpoint, ConnectionStatus.Awaiting); AddConnection(conn); return conn; } /// Start establishing a connection to a given endpoint with the given protocol public bool StartConnection(IPEndPoint endpoint, byte protocolIdent) { if (IPtoID.ContainsKey(endpoint)) { return false; } Connection conn = new Connection(endpoint); conn.Internal.AssignedProtocol = protocolIdent; AddConnection(conn); return true; } /// Get the connection instance from the given uid, if such exists. Null otherwise. public Connection GetConnection(ulong uid) { Connection conn; Connections.TryGetValue(uid, out conn); return conn; } /// Soft-closes the connection with the given uid, meaning it will wait for them to acknowledge the closing public void CloseConnection(ulong uid, ClosingReason reason) { if (Connections.ContainsKey(uid)) { Connections[uid].ClosingReason = reason; Connections[uid].Status = ConnectionStatus.Rejected; } } /// Add a reliable packet to the packet queue, to be sent on the next update, or when SendPacketQueue is called. public void AddPacketToQueue(ulong uid, Packet p) { if (!Connections.ContainsKey(uid)) { return; } p = p.ShallowCopy(); p.PacketId = Connections[uid].Internal.ReliablePacketIDCounter++; PacketQueue[uid].Add(p); } /// Send the current packet queue instantly. public void SendPacketQueue(ulong uid) { if (!Connections.ContainsKey(uid)) { return; } Connection conn = Connections[uid]; Protocol protocol = Peer.GetProtocol(conn.Internal.AssignedProtocol); if (protocol != null && conn.IsReady()) { ByteBuffer buffer = protocol.BuildMessage(conn); List list = PacketQueue[uid]; buffer.Write(list.Count); foreach (Packet p in list) { buffer.WritePacket(protocol, p); } Send(conn, buffer); } } /// Send a single unreliable packet. public void SendSingleUnreliable(ulong uid, Packet p) { if (!Connections.ContainsKey(uid)) { return; } Connection conn = Connections[uid]; p = p.ShallowCopy(); p.PacketId = conn.Internal.UnreliablePacketIDCounter++; p.PacketIsReliable = false; Protocol protocol = Peer.GetProtocol(conn.Internal.AssignedProtocol); if (protocol != null && conn.IsReady()) { ByteBuffer buffer = protocol.BuildMessage(conn); buffer.Write(1); buffer.WritePacket(protocol, p); Send(conn, buffer); } } /// Go through queue of networking actions that have happened since last update. public void Update() { foreach (byte id in ProtocolActionQueues.Keys) { Protocol protocol = Peer.GetProtocol(id); while (ProtocolActionQueues[id].Count > 0) { ProtocolAction action = ProtocolActionQueues[id].Dequeue(); if (action is ReceiveAction) { ReceiveAction receive = (ReceiveAction)action; protocol.Receive(receive.Connection, receive.Packet); } else if (action is ConnectionChangedAction) { ConnectionChangedAction changed = (ConnectionChangedAction)action; protocol.ConnectionStatusChanged(changed.OldStatus, changed.NewStatus, changed.Connection); } else if (action is TimeoutAction) { TimeoutAction changed = (TimeoutAction)action; protocol.Timeout(changed.Connection); } } } } private void AddConnection(Connection conn) { conn.uid = ConnectionCounter++; Connections.Add(conn.uid, conn); IPtoID.Add(conn.Endpoint, conn.uid); PacketQueue.Add(conn.uid, new List()); } private void RemoveConnection(Connection conn) { Connections.Remove(conn.uid); IPtoID.Remove(conn.Endpoint); PacketQueue.Remove(conn.uid); } private void SendPlain(Connection conn) { Protocol protocol = Peer.GetProtocol(conn.Internal.AssignedProtocol); if (protocol != null) { ByteBuffer buffer = protocol.BuildMessage(conn); Send(conn, buffer); } } private void Send(Connection conn, ByteBuffer buffer) { if (conn.Status == ConnectionStatus.Lost) { return; } byte[] bytes = buffer.Pack(); Peer.ListenerThread.LastSentConnection = conn; Peer.UdpClient.Send(bytes, bytes.Length, conn.Endpoint); } public void Handle(IPEndPoint endpoint, ByteBuffer buffer) { Connection conn = Find(endpoint); conn.Internal.LastMessage = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); ConnectionStatus oldStatus = conn.Status; byte protocolId = buffer.Read(); Protocol protocol = Peer.GetProtocol(protocolId); PacketStage stage = buffer.ReadStage(); switch (stage) { case PacketStage.Establishing: if (conn.Status == ConnectionStatus.Awaiting) { conn.Internal.AssignedProtocol = protocolId; string version = buffer.ReadString(); if (protocol == null || !version.Equals(protocol.Version)) { conn.Status = ConnectionStatus.Rejected; conn.ClosingReason = ClosingReason.IncorrectVersion; } else { conn.Status = ConnectionStatus.Ready; } if (protocol != null) { ProtocolActionQueues[protocol.Identifier].Enqueue(new ConnectionChangedAction(oldStatus, conn.Status, conn)); } } break; case PacketStage.Rejected: conn.Status = ConnectionStatus.Closed; conn.ClosingReason = buffer.ReadClosingReason(); if (protocol != null) { ProtocolActionQueues[protocol.Identifier].Enqueue(new ConnectionChangedAction(oldStatus, conn.Status, conn)); } break; case PacketStage.Closed: if (conn.Status == ConnectionStatus.Stopped) { break; } conn.Status = ConnectionStatus.Stopped; if (protocol != null) { ProtocolActionQueues[protocol.Identifier].Enqueue(new ConnectionChangedAction(oldStatus, conn.Status, conn)); } break; case PacketStage.Ready: if (conn.Internal.AssignedProtocol != protocolId || protocol == null) { break; } if (oldStatus == ConnectionStatus.Establishing) { // Update connection status conn.Status = ConnectionStatus.Ready; ProtocolActionQueues[protocol.Identifier].Enqueue(new ConnectionChangedAction(oldStatus, conn.Status, conn)); } if (!(oldStatus == ConnectionStatus.Establishing || oldStatus == ConnectionStatus.Ready)) { break; // No cheating at this table! For realsies this time! } conn.Internal.LatestOutwardReliable = buffer.ReadInt(); List list = PacketQueue[conn.uid]; list.RemoveAll(p => p.PacketId <= conn.Internal.LatestOutwardReliable); PacketQueue[conn.uid] = list; int PacketAmount = buffer.ReadInt(); for (int i = 0; i < PacketAmount; i++) { Packet p = buffer.ReadPacket(protocol); if (p.PacketIsReliable) { if (p.PacketId > conn.Internal.LatestInwardReliable) { conn.Internal.LatestInwardReliable = p.PacketId; ProtocolActionQueues[protocol.Identifier].Enqueue(new ReceiveAction(conn, p)); } } else if (p.PacketId > conn.Internal.LatestInwardUnreliable) { conn.Internal.LatestInwardUnreliable = p.PacketId; ProtocolActionQueues[protocol.Identifier].Enqueue(new ReceiveAction(conn, p)); } } break; } } private void UpdateThreadMethod() { try { while (Thread.CurrentThread.IsAlive) { long now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); List timedOut = new List(); foreach (ulong uid in Connections.Keys) { Connection conn = Connections[uid]; if ((now - conn.Internal.LastMessage) > Timeout || conn.Status == ConnectionStatus.Lost) { timedOut.Add(uid); } if (conn.Status != ConnectionStatus.Awaiting || conn.Status != ConnectionStatus.Stopped) { if (conn.Status == ConnectionStatus.Ready) { SendPacketQueue(uid); } else { SendPlain(conn); } } } foreach (ulong uid in timedOut) { Connection conn = Connections[uid]; RemoveConnection(conn); if (conn.Status == ConnectionStatus.Ready || conn.Status == ConnectionStatus.Establishing || conn.Status == ConnectionStatus.Awaiting || conn.Status == ConnectionStatus.Lost) { Protocol protocol = Peer.GetProtocol(conn.Internal.AssignedProtocol); if (protocol != null) { conn.ClosingReason = ClosingReason.Timeout; ProtocolActionQueues[conn.Internal.AssignedProtocol].Enqueue(new TimeoutAction(conn)); } } } Thread.Sleep((int)Interval); } } catch (ThreadAbortException) { Peer.MessageListener.Message("Connection Thread Stopped"); } } } public abstract class ProtocolAction { }; class ReceiveAction : ProtocolAction { public Connection Connection; public Packet Packet; public ReceiveAction(Connection connection, Packet packet) { Connection = connection; Packet = packet; } } class ConnectionChangedAction : ProtocolAction { public ConnectionStatus OldStatus; public ConnectionStatus NewStatus; public Connection Connection; public ConnectionChangedAction(ConnectionStatus old, ConnectionStatus newstatus, Connection connection) { Connection = connection; OldStatus = old; NewStatus = newstatus; } } class TimeoutAction : ProtocolAction { public Connection Connection; public TimeoutAction(Connection connection) { Connection = connection; } } }