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

356 lines
16 KiB
C#
Raw Normal View History

2020-08-05 03:21:04 +02:00
using System.Collections.Generic;
using System.Collections.Concurrent;
2020-08-05 03:21:04 +02:00
using System.Net;
using System.Threading;
using System;
using NeonTea.Quakeball.TeaNet.Packets;
using UnityEngine;
2020-08-05 03:21:04 +02:00
namespace NeonTea.Quakeball.TeaNet.Peers {
2020-08-06 20:49:51 +02:00
/// <summary>Manages connections for Peer, sends them keepalives and sends and handles incoming messages.</summary>
2020-08-05 03:21:04 +02:00
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, ConcurrentQueue<Packet>> PacketQueue = new Dictionary<ulong, ConcurrentQueue<Packet>>();
2020-08-05 03:21:04 +02:00
private Peer Peer;
public Dictionary<ulong, Queue<ProtocolAction>> ProtocolActionQueues = new Dictionary<ulong, Queue<ProtocolAction>>();
private Queue<Connection> ConnectionsToRemove = new Queue<Connection>();
2020-08-05 03:21:04 +02:00
private Thread UpdateThread;
public long Timeout = 8000;
2020-08-05 16:59:16 +02:00
public long Interval = 100;
2020-08-05 03:21:04 +02:00
/// <summary>The amount of bytes sent since the last clear interval from Peer</summary>
public int BytesSent;
public ConcurrentDictionary<Type, int> SentByPacket = new ConcurrentDictionary<Type, int>();
public ConcurrentDictionary<Type, int> ReceivedByPacket = new ConcurrentDictionary<Type, int>();
2020-08-05 03:21:04 +02:00
public ConnectionManager(Peer peer) {
Peer = peer;
UpdateThread = new Thread(new ThreadStart(UpdateThreadMethod));
UpdateThread.Start();
}
public void StopThread() {
UpdateThread.Abort();
}
2020-08-06 20:49:51 +02:00
/// <summary>Find a given Connection. Should not be used, unless expecting to establish a connection with the endpoint.</summary>
2020-08-05 03:21:04 +02:00
public Connection Find(IPEndPoint endpoint) {
if (IPtoID.ContainsKey(endpoint)) {
return Connections[IPtoID[endpoint]];
2020-08-05 03:21:04 +02:00
}
Connection conn = new Connection(endpoint, ConnectionStatus.Awaiting);
AddConnection(conn);
2020-08-05 03:21:04 +02:00
return conn;
}
2020-08-06 20:49:51 +02:00
/// <summary>Start establishing a connection to a given endpoint with the given protocol</summary>
2020-08-05 03:21:04 +02:00
public bool StartConnection(IPEndPoint endpoint, byte protocolIdent) {
if (IPtoID.ContainsKey(endpoint)) {
2020-08-05 03:21:04 +02:00
return false;
}
Connection conn = new Connection(endpoint);
conn.Internal.AssignedProtocol = protocolIdent;
AddConnection(conn);
2020-08-05 03:21:04 +02:00
return true;
}
2020-08-06 20:49:51 +02:00
/// <summary>Get the connection instance from the given uid, if such exists. Null otherwise.</summary>
public Connection GetConnection(ulong uid) {
Connection conn;
Connections.TryGetValue(uid, out conn);
return conn;
2020-08-05 03:21:04 +02:00
}
/// <summary>Soft-closes the connection with the given uid, meaning it will wait for them to acknowledge the closing</summary>
public void CloseConnection(ulong uid, ClosingReason reason) {
if (Connections.ContainsKey(uid)) {
Connections[uid].ClosingReason = reason;
Connections[uid].Status = ConnectionStatus.Rejected;
}
}
2020-08-06 20:49:51 +02:00
/// <summary>Add a reliable packet to the packet queue, to be sent on the next update, or when SendPacketQueue is called.</summary>
public void AddPacketToQueue(ulong uid, Packet p) {
if (!Connections.ContainsKey(uid)) {
return;
}
p = p.ShallowCopy();
p.PacketId = Connections[uid].Internal.ReliablePacketIDCounter++;
PacketQueue[uid].Enqueue(p);
}
2020-08-06 20:49:51 +02:00
/// <summary>Send the current packet queue instantly.</summary>
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()) {
Packet[] list = PacketQueue[uid].ToArray();
2020-08-10 23:20:58 +02:00
int firstId = Int32.MaxValue;
if (list.Length > 0) {
firstId = list[0].PacketId;
}
2020-08-11 00:05:04 +02:00
ByteBuffer buffer = protocol.BuildMessage(conn, firstId, true);
buffer.Write(list.Length);
2020-08-05 03:21:04 +02:00
foreach (Packet p in list) {
buffer.WritePacket(protocol, p);
// Do the analytics dance!
int OldSentByPacket;
SentByPacket.TryGetValue(p.GetType(), out OldSentByPacket);
SentByPacket[p.GetType()] = OldSentByPacket + p.Size;
2020-08-05 03:21:04 +02:00
}
Send(conn, buffer);
}
}
2020-08-06 20:49:51 +02:00
/// <summary>Send a single unreliable packet.</summary>
public void SendSingleUnreliable(ulong uid, Packet p) {
if (!Connections.ContainsKey(uid)) {
return;
}
Connection conn = Connections[uid];
2020-08-05 21:04:21 +02:00
p = p.ShallowCopy();
p.PacketId = conn.Internal.UnreliablePacketIDCounter++;
p.PacketIsReliable = false;
Protocol protocol = Peer.GetProtocol(conn.Internal.AssignedProtocol);
if (protocol != null && conn.IsReady()) {
2020-08-11 00:05:04 +02:00
ByteBuffer buffer = protocol.BuildMessage(conn, p.PacketId, false);
2020-08-05 20:38:37 +02:00
buffer.Write(1);
buffer.WritePacket(protocol, p);
2020-08-05 03:21:04 +02:00
Send(conn, buffer);
// Do the analytics dance!
int OldSentByPacket;
SentByPacket.TryGetValue(p.GetType(), out OldSentByPacket);
SentByPacket[p.GetType()] = OldSentByPacket + p.Size;
2020-08-05 03:21:04 +02:00
}
}
/// <summary>Go through queue of networking actions that have happened since last update.</summary>
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);
}
}
}
while (ConnectionsToRemove.Count > 0) {
Connection conn = ConnectionsToRemove.Dequeue();
RemoveConnection(conn);
}
}
private void AddConnection(Connection conn) {
conn.uid = ConnectionCounter++;
Connections.Add(conn.uid, conn);
IPtoID.Add(conn.Endpoint, conn.uid);
PacketQueue.Add(conn.uid, new ConcurrentQueue<Packet>());
}
private void RemoveConnection(Connection conn) {
Connections.Remove(conn.uid);
IPtoID.Remove(conn.Endpoint);
PacketQueue.Remove(conn.uid);
}
2020-08-05 03:21:04 +02:00
private void SendPlain(Connection conn) {
Protocol protocol = Peer.GetProtocol(conn.Internal.AssignedProtocol);
2020-08-05 03:21:04 +02:00
if (protocol != null) {
2020-08-11 00:05:04 +02:00
ByteBuffer buffer = protocol.BuildMessage(conn, Int32.MaxValue, false);
2020-08-05 03:21:04 +02:00
Send(conn, buffer);
}
}
private void Send(Connection conn, ByteBuffer buffer) {
if (conn.Status == ConnectionStatus.Lost) {
return;
}
BytesSent += buffer.Size;
2020-08-05 03:21:04 +02:00
byte[] bytes = buffer.Pack();
Peer.ListenerThread.LastSentConnection = conn;
Peer.UdpClient.Send(bytes, bytes.Length, conn.Endpoint);
}
public void Handle(IPEndPoint endpoint, ByteBuffer buffer) {
2020-08-05 03:21:04 +02:00
Connection conn = Find(endpoint);
conn.Internal.LastMessage = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
2020-08-05 03:21:04 +02:00
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;
2020-08-05 03:21:04 +02:00
string version = buffer.ReadString();
if (protocol == null || !version.Equals(protocol.Version)) {
conn.Status = ConnectionStatus.Rejected;
conn.ClosingReason = ClosingReason.IncorrectVersion;
2020-08-05 03:21:04 +02:00
} else {
conn.Status = ConnectionStatus.Ready;
}
if (protocol != null) {
ProtocolActionQueues[protocol.Identifier].Enqueue(new ConnectionChangedAction(oldStatus, conn.Status, conn));
2020-08-05 03:21:04 +02:00
}
}
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));
2020-08-05 03:21:04 +02:00
}
break;
case PacketStage.Closed:
if (conn.Status == ConnectionStatus.Stopped) {
break;
}
2020-08-05 03:21:04 +02:00
conn.Status = ConnectionStatus.Stopped;
if (protocol != null) {
ProtocolActionQueues[protocol.Identifier].Enqueue(new ConnectionChangedAction(oldStatus, conn.Status, conn));
2020-08-05 03:21:04 +02:00
}
break;
case PacketStage.Ready:
if (conn.Internal.AssignedProtocol != protocolId || protocol == null) {
2020-08-05 03:21:04 +02:00
break;
}
if (oldStatus == ConnectionStatus.Establishing) { // Update connection status
2020-08-05 03:21:04 +02:00
conn.Status = ConnectionStatus.Ready;
ProtocolActionQueues[protocol.Identifier].Enqueue(new ConnectionChangedAction(oldStatus, conn.Status, conn));
2020-08-05 03:21:04 +02:00
}
2020-08-07 04:34:38 +02:00
if (!(oldStatus == ConnectionStatus.Establishing || oldStatus == ConnectionStatus.Ready)) {
break; // No cheating at this table! For realsies this time!
}
conn.Internal.LatestOutwardReliable = buffer.ReadInt();
int FirstPacketId = buffer.ReadInt();
2020-08-11 00:05:04 +02:00
bool Reliable = buffer.ReadBool();
ConcurrentQueue<Packet> queue = PacketQueue[conn.uid];
Packet peeked;
while (queue.TryPeek(out peeked)) {
2020-08-10 23:20:58 +02:00
if (peeked.PacketId > conn.Internal.LatestOutwardReliable) {
break;
} else {
Packet LastRemoved;
queue.TryDequeue(out LastRemoved);
}
}
PacketQueue[conn.uid] = queue;
2020-08-05 03:21:04 +02:00
int PacketAmount = buffer.ReadInt();
for (int i = 0; i < PacketAmount; i++) {
Packet p = buffer.ReadPacket(protocol);
2020-08-10 23:20:58 +02:00
p.PacketId = FirstPacketId + i;
2020-08-11 00:05:04 +02:00
p.PacketIsReliable = Reliable;
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));
}
// Do some analytics!
int OldReceivedByPacket;
ReceivedByPacket.TryGetValue(p.GetType(), out OldReceivedByPacket);
ReceivedByPacket[p.GetType()] = OldReceivedByPacket + p.Size;
2020-08-05 03:21:04 +02:00
}
break;
}
}
private void UpdateThreadMethod() {
try {
2020-08-05 16:59:16 +02:00
while (Thread.CurrentThread.IsAlive) {
2020-08-05 03:21:04 +02:00
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);
2020-08-05 16:59:16 +02:00
}
if (conn.Status != ConnectionStatus.Awaiting || conn.Status != ConnectionStatus.Stopped) {
if (conn.Status == ConnectionStatus.Ready) {
SendPacketQueue(uid);
2020-08-05 16:59:16 +02:00
} else {
SendPlain(conn);
2020-08-05 03:21:04 +02:00
}
}
2020-08-05 16:59:16 +02:00
}
foreach (ulong uid in timedOut) {
Connection conn = Connections[uid];
ConnectionsToRemove.Enqueue(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);
2020-08-05 16:59:16 +02:00
if (protocol != null) {
conn.ClosingReason = ClosingReason.Timeout;
ProtocolActionQueues[conn.Internal.AssignedProtocol].Enqueue(new TimeoutAction(conn));
2020-08-05 03:21:04 +02:00
}
}
}
2020-08-05 16:59:16 +02:00
Thread.Sleep((int)Interval);
2020-08-05 03:21:04 +02:00
}
2020-08-05 16:59:16 +02:00
} catch (ThreadAbortException) {
Peer.MessageListener.Message("Connection Thread Stopped");
2020-08-05 03:21:04 +02:00
}
}
}
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;
}
}
2020-08-05 03:21:04 +02:00
}