Add fragmentation support

Hazel.UnitTests/FragmentationTests.cs

@@ -0,0 +1,80 @@
+using System;
+using System.Linq;
+using System.Net;
+using System.Threading;
+using Hazel.Udp;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace Hazel.UnitTests
+{
+    [TestClass]
+    public class FragmentationTests
+    {
+        private readonly byte[] _testData = Enumerable.Range(0, 10000).Select(x => (byte)x).ToArray();
+
+        [TestMethod]
+        [DataRow(false, DisplayName = "SendBytes")]
+        [DataRow(true, DisplayName = "MessageWriter")]
+        public void ReliableSendTest(bool useMessageWriter)
+        {
+            using (var listener = new UdpConnectionListener(new IPEndPoint(IPAddress.Any, 4296)))
+            using (var connection = new UdpClientConnection(new IPEndPoint(IPAddress.Loopback, 4296)))
+            {
+                var manualResetEvent = new ManualResetEventSlim(false);
+                DataReceivedEventArgs? data = null;
+
+                listener.NewConnection += e =>
+                {
+                    e.Connection.DataReceived += de =>
+                    {
+                        data = de;
+                        manualResetEvent.Set();
+                    };
+                };
+
+                listener.Start();
+                connection.Connect();
+
+                if (useMessageWriter)
+                {
+                    var messageWriter = MessageWriter.Get(SendOption.Reliable);
+                    messageWriter.Write(_testData);
+                    connection.Send(messageWriter);
+                }
+                else
+                {
+                    connection.SendBytes(_testData, SendOption.Reliable);
+                }
+
+                manualResetEvent.Wait(TimeSpan.FromSeconds(5));
+
+                Assert.IsNotNull(data);
+
+                Assert.AreEqual(SendOption.Reliable, data.Value.SendOption);
+
+                var messageReader = data.Value.Message;
+                var received = new byte[messageReader.Length];
+                Array.Copy(messageReader.Buffer, messageReader.Offset + messageReader.Position, received, 0, received.Length);
+                messageReader.Recycle();
+
+                CollectionAssert.AreEqual(_testData, received);
+            }
+        }
+
+        [TestMethod]
+        public void UnreliableSendTest()
+        {
+            using (var listener = new UdpConnectionListener(new IPEndPoint(IPAddress.Any, 4296)))
+            using (var connection = new UdpClientConnection(new IPEndPoint(IPAddress.Loopback, 4296)))
+            {
+                listener.Start();
+                connection.Connect();
+
+                Assert.AreEqual("Unreliable messages can't be bigger than MTU", Assert.ThrowsException<HazelException>(() =>
+                {
+                    connection.SendBytes(_testData);
+                }).Message);
+            }
+        }
+    }
+}

Hazel.UnitTests/UdpConnectionTestHarness.cs

@@ -38,7 +38,7 @@ protected override bool SendDisconnect(MessageWriter writer)
             return true;
         }
 
-        protected override void WriteBytesToConnection(byte[] bytes, int length)
+        protected override void WriteBytesToConnection(byte[] bytes, int length, Action onTooBig = null)
         {
             this.BytesSent.Add(MessageReader.Get(bytes));
         }

Hazel/Dtls/DtlsConnectionListener.cs

@@ -1258,7 +1258,7 @@ private void SendHelloVerifyRequest(IPEndPoint peerAddress, ulong recordSequence
         /// <summary>
         /// Handle a requrest to send a datagram to the network
         /// </summary>
-        protected override void QueueRawData(ByteSpan span, IPEndPoint remoteEndPoint)
+        protected override void QueueRawData(ByteSpan span, IPEndPoint remoteEndPoint, Action onTooBig = null)
         {
             PeerData peer;
             if (!this.existingPeers.TryGetValue(remoteEndPoint, out peer))

Hazel/Dtls/DtlsUnityConnection.cs

@@ -325,13 +325,13 @@ private ByteSpan WriteBytesToConnectionInternal(byte[] bytes, int length)
         }
 
         /// <inheritdoc />
-        protected override void WriteBytesToConnection(byte[] bytes, int length)
+        protected override void WriteBytesToConnection(byte[] bytes, int length, Action onTooBig = null)
         {
             ByteSpan wireData = this.WriteBytesToConnectionInternal(bytes, length);
             if (wireData.Length > 0)
             {
                 Debug.Assert(wireData.Offset ==  0, "Got a non-zero write data offset");
-                base.WriteBytesToConnection(wireData.GetUnderlyingArray(), wireData.Length);
+                base.WriteBytesToConnection(wireData.GetUnderlyingArray(), wireData.Length, onTooBig);
             }
         }
 

Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs

@@ -17,6 +17,7 @@ private struct SendMessageInfo
         {
             public ByteSpan Span;
             public IPEndPoint Recipient;
+            public Action OnTooBig;
         }
 
         private struct ReceiveMessageInfo
@@ -250,7 +251,14 @@ private void SendLoop()
                 {
                     if (this.socket.Poll(Timeout.Infinite, SelectMode.SelectWrite))
                     {
-                        this.socket.SendTo(msg.Span.GetUnderlyingArray(), msg.Span.Offset, msg.Span.Length, SocketFlags.None, msg.Recipient);
+                        try
+                        {
+                            this.socket.SendTo(msg.Span.GetUnderlyingArray(), msg.Span.Offset, msg.Span.Length, SocketFlags.None, msg.Recipient);
+                        }
+                        catch (SocketException e) when (msg.OnTooBig != null && e.SocketErrorCode == SocketError.MessageSize)
+                        {
+                            msg.OnTooBig();
+                        }
                     }
                     else
                     {
@@ -335,14 +343,14 @@ protected virtual void ReadCallback(MessageReader message, IPEndPoint remoteEndP
             connection.HandleReceive(message, bytesReceived);
         }
 
-        internal void SendDataRaw(byte[] response, IPEndPoint remoteEndPoint)
+        internal void SendDataRaw(byte[] response, IPEndPoint remoteEndPoint, Action onTooBig = null)
         {
-            QueueRawData(response, remoteEndPoint);
+            QueueRawData(response, remoteEndPoint, onTooBig);
         }
 
-        protected virtual void QueueRawData(ByteSpan span, IPEndPoint remoteEndPoint)
+        protected virtual void QueueRawData(ByteSpan span, IPEndPoint remoteEndPoint, Action onTooBig = null)
         {
-            this.sendQueue.TryAdd(new SendMessageInfo() { Span = span, Recipient = remoteEndPoint });
+            this.sendQueue.TryAdd(new SendMessageInfo() { Span = span, Recipient = remoteEndPoint, OnTooBig = onTooBig });
         }
 
         /// <summary>

Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs

@@ -38,14 +38,15 @@ internal ThreadLimitedUdpServerConnection(ThreadLimitedUdpConnectionListener lis
 
             State = ConnectionState.Connected;
             this.InitializeKeepAliveTimer();
+            this.StartMtuDiscovery();
         }
 
         /// <inheritdoc />
-        protected override void WriteBytesToConnection(byte[] bytes, int length)
+        protected override void WriteBytesToConnection(byte[] bytes, int length, Action onTooBig = null)
         {
             if (bytes.Length != length) throw new ArgumentException("I made an assumption here. I hope you see this error.");
 
-            Listener.SendDataRaw(bytes, EndPoint);
+            Listener.SendDataRaw(bytes, EndPoint, onTooBig);
         }
 
         /// <inheritdoc />

Hazel/MessageWriter.cs

@@ -33,10 +33,12 @@ public MessageWriter(int bufferSize)
 
         public byte[] ToByteArray(bool includeHeader)
         {
+            var length = GetLength(includeHeader);
+
             if (includeHeader)
             {
-                byte[] output = new byte[this.Length];
-                System.Buffer.BlockCopy(this.Buffer, 0, output, 0, this.Length);
+                byte[] output = new byte[length];
+                System.Buffer.BlockCopy(this.Buffer, 0, output, 0, length);
                 return output;
             }
             else
@@ -45,14 +47,14 @@ public byte[] ToByteArray(bool includeHeader)
                 {
                     case SendOption.Reliable:
                         {
-                            byte[] output = new byte[this.Length - 3];
-                            System.Buffer.BlockCopy(this.Buffer, 3, output, 0, this.Length - 3);
+                            byte[] output = new byte[length];
+                            System.Buffer.BlockCopy(this.Buffer, 3, output, 0, length);
                             return output;
                         }
                     case SendOption.None:
                         {
-                            byte[] output = new byte[this.Length - 1];
-                            System.Buffer.BlockCopy(this.Buffer, 1, output, 0, this.Length - 1);
+                            byte[] output = new byte[length];
+                            System.Buffer.BlockCopy(this.Buffer, 1, output, 0, length);
                             return output;
                         }
                 }
@@ -61,6 +63,28 @@ public byte[] ToByteArray(bool includeHeader)
             throw new NotImplementedException();
         }
 
+        public int GetLength(bool includeHeader)
+        {
+            if (includeHeader)
+            {
+                return Length;
+            }
+
+            switch (this.SendOption)
+            {
+                case SendOption.Reliable:
+                {
+                    return Length - 3;
+                }
+                case SendOption.None:
+                {
+                    return Length - 1;
+                }
+            }
+
+            throw new NotImplementedException();
+        }
+
         ///
         /// <param name="sendOption">The option specifying how the message should be sent.</param>
         public static MessageWriter Get(SendOption sendOption = SendOption.None)

Hazel/NetworkConnection.cs

@@ -14,7 +14,8 @@ public enum HazelInternalErrors
         ReceivedZeroBytes,
         PingsWithoutResponse,
         ReliablePacketWithoutResponse,
-        ConnectionDisconnected
+        ConnectionDisconnected,
+        MtuTooLow
     }
 
     /// <summary>

Hazel/Udp/SendOptionInternal.cs

@@ -35,5 +35,10 @@ public enum UdpSendOption : byte
         ///     Message that is part of a larger, fragmented message.
         /// </summary>
         Fragment = 11,
+
+        /// <summary>
+        ///     Message that is used to discover MTU.
+        /// </summary>
+        MtuTest = 13,
     }
 }

Hazel/Udp/UdpClientConnection.cs

@@ -61,21 +61,21 @@ private void ManageReliablePacketsInternal(object state)
         }
 
         /// <inheritdoc />
-        protected override void WriteBytesToConnection(byte[] bytes, int length)
+        protected override void WriteBytesToConnection(byte[] bytes, int length, Action onTooBig = null)
         {
 #if DEBUG
             if (TestLagMs > 0)
             {
-                ThreadPool.QueueUserWorkItem(a => { Thread.Sleep(this.TestLagMs); WriteBytesToConnectionReal(bytes, length); });
+                ThreadPool.QueueUserWorkItem(a => { Thread.Sleep(this.TestLagMs); WriteBytesToConnectionReal(bytes, length, onTooBig); });
             }
             else
 #endif
             {
-                WriteBytesToConnectionReal(bytes, length);
+                WriteBytesToConnectionReal(bytes, length, onTooBig);
             }
         }
 
-        private void WriteBytesToConnectionReal(byte[] bytes, int length)
+        private void WriteBytesToConnectionReal(byte[] bytes, int length, Action onTooBig)
         {
 #if DEBUG
             DataSentRaw?.Invoke(bytes, length);
@@ -97,6 +97,10 @@ private void WriteBytesToConnectionReal(byte[] bytes, int length)
             {
                 // Already disposed and disconnected...
             }
+            catch (SocketException e) when (onTooBig != null && e.SocketErrorCode == SocketError.MessageSize)
+            {
+                onTooBig();
+            }
             catch (SocketException ex)
             {
                 DisconnectInternal(HazelInternalErrors.SocketExceptionSend, "Could not send data as a SocketException occurred: " + ex.Message);
@@ -114,6 +118,10 @@ private void HandleSendTo(IAsyncResult result)
             {
                 // Already disposed and disconnected...
             }
+            catch (SocketException e) when (e.SocketErrorCode == SocketError.MessageSize)
+            {
+                throw;
+            }
             catch (SocketException ex)
             {
                 DisconnectInternal(HazelInternalErrors.SocketExceptionSend, "Could not send data as a SocketException occurred: " + ex.Message);
@@ -177,6 +185,7 @@ public override void ConnectAsync(byte[] bytes = null)
             {
                 this.State = ConnectionState.Connected;
                 this.InitializeKeepAliveTimer();
+                this.StartMtuDiscovery();
             });
         }