quakeball/Assets/Scripts/TeaNet/Peers/ConnectionManager.cs

235 lines
9.7 KiB
C#

using UnityEngine;
using System.Collections.Generic;
using System.Net;
using System.Threading;
using System;
using NeonTea.Quakeball.TeaNet.Packets;
namespace NeonTea.Quakeball.TeaNet.Peers {
public class ConnectionManager {
private ulong ConnectionCounter;
private Dictionary<ulong, Connection> Connections = new Dictionary<ulong, Connection>();
private Dictionary<IPEndPoint, ulong> IPtoID = new Dictionary<IPEndPoint, ulong>();
private Dictionary<ulong, List<Packet>> PacketQueue = new Dictionary<ulong, List<Packet>>();
private Peer Peer;
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();
}
public Connection Find(IPEndPoint endpoint) {
if (IPtoID.ContainsKey(endpoint)) {
return Connections[IPtoID[endpoint]];
}
Connection conn = new Connection(endpoint, ConnectionStatus.Awaiting);
AddConnection(conn);
return conn;
}
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;
}
public Connection GetConnection(ulong uid) {
Connection conn;
Connections.TryGetValue(uid, out conn);
return conn;
}
public void AddPacketToQueue(ulong uid, Packet p) {
if (!Connections.ContainsKey(uid)) {
return;
}
p.Id = Connections[uid].Internal.ReliablePacketIDCounter++;
PacketQueue[uid].Add(p);
}
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<Packet> list = PacketQueue[uid];
buffer.WriteInt(list.Count);
foreach (Packet p in list) {
buffer.WritePacket(protocol, p);
}
Send(conn, buffer);
}
}
public void SendSingleUnreliable(ulong uid, Packet p) {
if (!Connections.ContainsKey(uid)) {
return;
}
Connection conn = Connections[uid];
p.Id = conn.Internal.UnreliablePacketIDCounter++;
p.Reliable = false;
Protocol protocol = Peer.GetProtocol(conn.Internal.AssignedProtocol);
if (protocol != null && conn.IsReady()) {
ByteBuffer buffer = protocol.BuildMessage(conn);
buffer.WriteInt(1);
buffer.WritePacket(protocol, p);
Send(conn, buffer);
}
}
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<Packet>());
}
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) {
protocol.ConnectionStatusChanged(oldStatus, conn.Status, conn);
}
}
break;
case PacketStage.Rejected:
conn.Status = ConnectionStatus.Closed;
conn.ClosingReason = buffer.ReadClosingReason();
if (protocol != null) {
protocol.ConnectionStatusChanged(oldStatus, conn.Status, conn);
}
break;
case PacketStage.Closed:
if (conn.Status == ConnectionStatus.Stopped) {
break;
}
conn.Status = ConnectionStatus.Stopped;
if (protocol != null) {
protocol.ConnectionStatusChanged(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;
protocol.ConnectionStatusChanged(oldStatus, conn.Status, conn);
}
conn.Internal.LatestOutwardReliable = buffer.ReadInt();
List<Packet> list = PacketQueue[conn.uid];
list.RemoveAll(p => p.Id <= 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.Reliable) {
if (p.Id > conn.Internal.LatestInwardReliable) {
conn.Internal.LatestInwardReliable = p.Id;
protocol.Receive(conn, p);
}
} else if (p.Id > conn.Internal.LatestInwardUnreliable) {
conn.Internal.LatestInwardUnreliable = p.Id;
protocol.Receive(conn, p);
}
}
break;
}
}
private void UpdateThreadMethod() {
try {
while (Thread.CurrentThread.IsAlive) {
long now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
List<ulong> timedOut = new List<ulong>();
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) {
protocol.Timeout(conn);
}
}
}
Thread.Sleep((int)Interval);
}
} catch (ThreadAbortException) {
Peer.MessageListener.Message("Connection Thread Stopped");
}
}
}
}