public event Action<DataReceivedEventArgs> DataReceived;
public int TestLagMs = -1;
+ public int TestDropRate = 0;
+ protected int testDropCount = 0;
/// <summary>
/// Called when the end point disconnects or an error occurs.
}
case SendOption.Tcp:
{
- byte[] output = new byte[this.Length - 4];
- System.Buffer.BlockCopy(this.Buffer, 4, output, 0, this.Length - 4);
+ byte[] output = new byte[this.Length];
+ System.Buffer.BlockCopy(this.Buffer, 0, output, 0, this.Length);
return output;
}
}
public void Clear(SendOption sendOption)
{
- this.Position = this.Length = 0;
+ this.messageStarts.Clear();
this.SendOption = sendOption;
-
this.Buffer[0] = (byte)sendOption;
switch (sendOption)
{
this.Length = this.Position = 3;
break;
case SendOption.Tcp:
+ this.Length = this.Position = 0;
break;
}
}
//Start receiving data
try
{
- ListenForData(InvokeAndListen);
+ var msg = MessageReader.GetSized(ushort.MaxValue);
+
+ ListenForData(msg, InvokeAndListen);
}
catch (Exception e)
{
{
this.State = ConnectionState.Connected;
+ var buffer = MessageReader.GetSized(ushort.MaxValue);
try
{
- ListenForData(
- delegate (MessageReader msg)
- {
- ListenForData(InvokeAndListen);
-
- //Remove version byte
- msg.Offset = 1;
- msg.Length -= 1;
- msg.Position = 0;
+ buffer.Offset = 0;
+ buffer.Length = 4;
+ buffer.Position = 0;
- callback.Invoke(msg);
- }
+ ListenForData(
+ buffer,
+ m => ReadHeader(m,
+ delegate (MessageReader msg)
+ {
+ ListenForData();
+
+ //Remove version byte
+ msg.Offset = 1;
+ msg.Length -= 1;
+ msg.Position = 0;
+
+ callback.Invoke(msg);
+ })
);
}
catch (Exception e)
{
+ buffer.Recycle();
Disconnect("An exception occured while initiating the first receive operation: " + e.Message);
}
}
private void InvokeAndListen(MessageReader msg)
{
- this.ListenForData(InvokeAndListen);
+ this.ListenForData();
try
{
}
catch { }
}
-
- private void ListenForData(Action<MessageReader> callback)
+
+ private void ListenForData()
{
- if (State == ConnectionState.Disconnecting || State == ConnectionState.NotConnected)
- throw new HazelException("Not connected");
-
var msg = MessageReader.GetSized(ushort.MaxValue);
- try
- {
- socket.BeginReceive(msg.Buffer, 0, 4, SocketFlags.None, o => HeaderReadCallback(callback, o), msg);
- }
- catch (SocketException s)
- {
- Disconnect("SocketException while reading header: " + s.Message);
- }
+ msg.Offset = 0;
+ msg.Length = 4;
+ msg.Position = 0;
+
+ ListenForData(msg, m => ReadHeader(m, null));
}
- private void HeaderReadCallback(Action<MessageReader> callback, IAsyncResult result)
+ private void ReadHeader(MessageReader msg, Action<MessageReader> callback)
{
- int bytesRead;
- try
- {
- bytesRead = socket.EndReceive(result);
- if (bytesRead == 0)
- {
- Disconnect("Received 0 bytes");
- return;
- }
-
- Statistics.LogFragmentedReceive(0, bytesRead);
- }
- catch (SocketException s)
- {
- Disconnect("SocketException while reading header: " + s.Message);
- return;
- }
-
- // TODO: Could possibly fragment here...
- var msg = (MessageReader)result.AsyncState;
msg.Length = GetLengthFromBytes(msg.Buffer);
+ msg.Position = 0;
+
+ ListenForData(msg, callback ?? InvokeAndListen);
+ }
+ private void ListenForData(MessageReader msg, Action<MessageReader> callback)
+ {
+ if (State == ConnectionState.Disconnecting || State == ConnectionState.NotConnected)
+ throw new HazelException("Not connected");
+
try
{
- socket.BeginReceive(msg.Buffer, 0, msg.Length, SocketFlags.None, o => BodyReadCallback(callback, o), msg);
+ socket.BeginReceive(msg.Buffer, msg.Position, msg.Length, SocketFlags.None, o => ReadUntilFull(callback, o), msg);
}
catch (SocketException s)
{
- Disconnect("SocketException while reading body: " + s.Message);
+ msg.Recycle();
+ Disconnect("SocketException while reading header: " + s.Message);
}
}
-
- private void BodyReadCallback(Action<MessageReader> callback, IAsyncResult result)
+
+ private void ReadUntilFull(Action<MessageReader> callback, IAsyncResult result)
{
int bytesRead;
try
if (msg.Position < bytesRead)
{
- try
- {
- socket.BeginReceive(msg.Buffer, msg.Position, msg.Length - msg.Position, SocketFlags.None, o => BodyReadCallback(callback, o), msg);
- }
- catch (SocketException s)
- {
- Disconnect("SocketException while reading body: " + s.Message);
- }
+ ListenForData(msg, callback);
}
else
{
}
IPEndPoint ipEnd = (IPEndPoint)endpt;
- string data = ASCIIEncoding.ASCII.GetString(buffer, 2, numBytes - 2);
+ string data = UTF8Encoding.UTF8.GetString(buffer, 2, numBytes - 2);
int dataHash = data.GetHashCode();
lock (packets)
{
if (this.socket != null)
{
- this.socket.Close();
+ try
+ {
+ this.socket.Shutdown(SocketShutdown.Both);
+ }
+ catch { }
+ try
+ {
+ this.socket.Close();
+ }
+ catch { }
+ try
+ {
+ this.socket.Dispose();
+ }
+ catch { }
this.socket = null;
}
}
///
public void SetData(string data)
{
- int len = ASCIIEncoding.ASCII.GetByteCount(data);
+ int len = UTF8Encoding.UTF8.GetByteCount(data);
this.data = new byte[len + 2];
this.data[0] = 4;
this.data[1] = 2;
- ASCIIEncoding.ASCII.GetBytes(data, 0, data.Length, this.data, 2);
+ UTF8Encoding.UTF8.GetBytes(data, 0, data.Length, this.data, 2);
}
///
{
if (this.socket != null)
{
- this.socket.Close();
+ try
+ {
+ this.socket.Shutdown(SocketShutdown.Both);
+ }
+ catch { }
+ try
+ {
+ this.socket.Close();
+ }
+ catch { }
+ try
+ {
+ this.socket.Dispose();
+ }
+ catch { }
this.socket = null;
}
}
/// <summary>
/// The socket we're connected via.
/// </summary>
- Socket socket;
+ private Socket socket;
/// <summary>
/// The buffer to store incomming data in.
/// </summary>
- byte[] dataBuffer = new byte[ushort.MaxValue];
+ private byte[] dataBuffer = new byte[ushort.MaxValue];
- Timer reliablePacketTimer;
+ private Timer reliablePacketTimer;
/// <summary>
/// Creates a new UdpClientConnection.
}
}
-
public event Action<byte[], int> DataSentRaw;
public event Action<byte[]> DataReceivedRaw;
Thread.Sleep(this.TestLagMs);
}
+ if (this.TestDropRate > 0)
+ {
+ if ((this.testDropCount++ % this.TestDropRate) == 0)
+ {
+ return;
+ }
+ }
+
DataReceivedRaw?.Invoke(bytes);
MessageReader msg = MessageReader.GetRaw(bytes, 0, bytesReceived);
HandleReceive(msg, bytesReceived);
/// resulting in a more dynamic resend that responds to endpoints on slower or faster connections.
/// </para>
/// </remarks>
- public int ResendTimeout { get { return resendTimeout; } set { resendTimeout = value; } }
- private volatile int resendTimeout = 0;
+ public volatile int ResendTimeout = 0;
+ /// <summary>
+ /// Max number of times to resend. 0 == no limit
+ /// </summary>
public volatile int ResendLimit = 0;
+ /// <summary>
+ /// A compounding multiplier to back off resend timeout.
+ /// Applied to ping before first timeout when ResendTimeout == 0.
+ /// </summary>
public volatile float ResendPingMultiplier = 3;
/// <summary>
/// <summary>
/// The packets of data that have been transmitted reliably and not acknowledged.
/// </summary>
- ConcurrentDictionary<ushort, Packet> reliableDataPacketsSent = new ConcurrentDictionary<ushort, Packet>();
+ internal ConcurrentDictionary<ushort, Packet> reliableDataPacketsSent = new ConcurrentDictionary<ushort, Packet>();
/// <summary>
/// The last packets that were received.
/// </summary>
- HashSet<ushort> reliableDataPacketsMissing = new HashSet<ushort>();
+ private HashSet<ushort> reliableDataPacketsMissing = new HashSet<ushort>();
/// <summary>
/// The packet id that was received last.
/// </summary>
- volatile ushort reliableReceiveLast = 0;
-
+ private volatile ushort reliableReceiveLast = 0;
+
/// <summary>
/// Has the connection received anything yet
/// </summary>
- volatile bool hasReceivedSomething = false;
+ private volatile bool hasReceivedSomething = false;
- object PingLock = new object();
+ private object PingLock = new object();
/// <summary>
/// Returns the average ping to this endpoint.
/// connection will be marked as disconnected and the <see cref="Connection.Disconnected">Disconnected</see> event
/// will be invoked.
/// </remarks>
- public volatile int DisconnectTimeout = 2500;
+ public volatile int DisconnectTimeout = 5000;
/// <summary>
/// Class to hold packet data
public int Retransmissions;
public Stopwatch Stopwatch = new Stopwatch();
-
+
Packet()
{
}
-
+
internal void Set(ushort id, UdpConnection connection, byte[] data, int length, int timeout, Action ackCallback)
{
this.Id = id;
if (lifetime >= this.NextTimeout)
{
- if (++this.Retransmissions > connection.ResendLimit)
+ ++this.Retransmissions;
+ if (connection.ResendLimit != 0
+ && this.Retransmissions > connection.ResendLimit)
{
if (connection.reliableDataPacketsSent.TryRemove(this.Id, out Packet self))
{
return 0;
}
- this.NextTimeout = (int)Math.Min(this.NextTimeout * 3f, connection.DisconnectTimeout);
+ this.NextTimeout = (int)Math.Min(this.NextTimeout * connection.ResendPingMultiplier, connection.DisconnectTimeout);
try
{
connection.WriteBytesToConnection(this.Data, this.Length);
PacketPool.PutObject(this);
}
}
-
+
internal int ManageReliablePackets()
{
int output = 0;
if (this.reliableDataPacketsSent.Count > 0)
{
- double minTimeout = int.MaxValue;
foreach (var kvp in this.reliableDataPacketsSent)
{
Packet pkt = kvp.Value;
output += pkt.Resend();
}
catch { }
-
- minTimeout = Math.Min(pkt.NextTimeout, minTimeout);
}
}
this,
buffer,
sendLength,
- resendTimeout > 0 ? resendTimeout : (int)Math.Max(300, Math.Min(AveragePingMs * this.ResendPingMultiplier, 2000)),
+ ResendTimeout > 0 ? ResendTimeout : (int)Math.Max(300, Math.Min(AveragePingMs * this.ResendPingMultiplier, 2000)),
ackCallback);
if (!reliableDataPacketsSent.TryAdd(id, packet))
{
this.Dispose(false);
}
-
- public float AveragePacketsTime = 1;
-
- Stopwatch stopwatch = new Stopwatch();
+
private void ManageReliablePackets(object state)
{
- stopwatch.Restart();
foreach (var kvp in this.allConnections)
{
var sock = kvp.Value;
sock.ManageReliablePackets();
}
-
- this.AveragePacketsTime = this.AveragePacketsTime * .7f + stopwatch.ElapsedMilliseconds * .3f;
-
+
this.reliablePacketTimer.Change(100, Timeout.Infinite);
}
if (this.socket != null)
{
+ try
+ {
+ this.socket.Shutdown(SocketShutdown.Both);
+ }
+ catch { }
this.socket.Close();
this.socket.Dispose();
this.socket = null;
projects / rhonda / impostor.hazel.git / commitdiff