Use abstract connection identifier

This change lays the groundwork for the
ThreadLimitedUdpConnectionListener to distinguish different logical
sessions from the same remote address.

diff --git a/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs b/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs
index 4ebff2e170a60620b47b3293edbae8307982686e..dcb42689d327a1c3958909c54a9333ba8f4fc23c 100644 (file)
--- a/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs
+++ b/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs
@@ -23,6 +23,7 @@ namespace Hazel.Udp.FewerThreads
         {
             public MessageReader Message;
             public EndPoint Sender;
+            public ConnectionId ConnectionId;
         }
 
         private const int SendReceiveBufferSize = 1024 * 1024;
@@ -49,7 +50,31 @@ namespace Hazel.Udp.FewerThreads
         private Thread sendThread;
         private HazelThreadPool processThreads;
 
-        private ConcurrentDictionary<EndPoint, ThreadLimitedUdpServerConnection> allConnections = new ConcurrentDictionary<EndPoint, ThreadLimitedUdpServerConnection>();
+        public struct ConnectionId
+        {
+            public ulong Id;
+
+            public static ConnectionId Create(ulong id)
+            {
+                ConnectionId result = new ConnectionId();
+                result.Id = id;
+                return result;
+            }
+
+            public static ConnectionId CreateFromEndPoint(IPEndPoint endPoint)
+            {
+                if (endPoint.AddressFamily != AddressFamily.InterNetwork)
+                {
+                    throw new ArgumentException("ConnectionId only supports IPv4");
+                }
+
+                ulong port = (ulong)endPoint.Port;
+                ulong address = (ulong)endPoint.Address.Address;
+                return Create((address << 32) | port);
+            }
+        }
+
+        private ConcurrentDictionary<ulong, ThreadLimitedUdpServerConnection> allConnections = new ConcurrentDictionary<ulong, ThreadLimitedUdpServerConnection>();
 
         private BlockingCollection<ReceiveMessageInfo> receiveQueue;
         private BlockingCollection<SendMessageInfo> sendQueue = new BlockingCollection<SendMessageInfo>();
@@ -171,7 +196,8 @@ namespace Hazel.Udp.FewerThreads
                         return;
                     }
 
-                    this.ProcessIncomingMessageFromOtherThread(message, remoteEP);
+                    ConnectionId connectionId = ConnectionId.CreateFromEndPoint((IPEndPoint)remoteEP);
+                    this.ProcessIncomingMessageFromOtherThread(message, remoteEP, connectionId);
                 }
             }
         }
@@ -183,7 +209,7 @@ namespace Hazel.Udp.FewerThreads
                 
                 try
                 {
-                    this.ReadCallback(msg.Message, msg.Sender);
+                    this.ReadCallback(msg.Message, msg.Sender, msg.ConnectionId);
                 }
                 catch
                 {
@@ -191,9 +217,9 @@ namespace Hazel.Udp.FewerThreads
                 }
             }
         }
-        protected virtual void ProcessIncomingMessageFromOtherThread(MessageReader message, EndPoint peerAddress)
+        protected virtual void ProcessIncomingMessageFromOtherThread(MessageReader message, EndPoint remoteEndPoint, ConnectionId connectionId)
         {
-            this.receiveQueue.Add(new ReceiveMessageInfo() { Message = message, Sender = peerAddress });
+            this.receiveQueue.Add(new ReceiveMessageInfo() { Message = message, Sender = remoteEndPoint, ConnectionId = connectionId });
         }
 
         private void SendLoop()
@@ -217,7 +243,7 @@ namespace Hazel.Udp.FewerThreads
             }
         }
 
-        void ReadCallback(MessageReader message, EndPoint remoteEndPoint)
+        void ReadCallback(MessageReader message, EndPoint remoteEndPoint, ConnectionId connectionId)
         {
             int bytesReceived = message.Length;
             bool aware = true;
@@ -226,11 +252,11 @@ namespace Hazel.Udp.FewerThreads
             // If we're aware of this connection use the one already
             // If this is a new client then connect with them!
             ThreadLimitedUdpServerConnection connection;
-            if (!this.allConnections.TryGetValue(remoteEndPoint, out connection))
+            if (!this.allConnections.TryGetValue(connectionId.Id, out connection))
             {
                 lock (this.allConnections)
                 {
-                    if (!this.allConnections.TryGetValue(remoteEndPoint, out connection))
+                    if (!this.allConnections.TryGetValue(connectionId.Id, out connection))
                     {
                         // Check for malformed connection attempts
                         if (!isHello)
@@ -254,8 +280,8 @@ namespace Hazel.Udp.FewerThreads
                         }
 
                         aware = false;
-                        connection = new ThreadLimitedUdpServerConnection(this, (IPEndPoint)remoteEndPoint, this.IPMode);
-                        if (!this.allConnections.TryAdd(remoteEndPoint, connection))
+                        connection = new ThreadLimitedUdpServerConnection(this, connectionId, (IPEndPoint)remoteEndPoint, this.IPMode);
+                        if (!this.allConnections.TryAdd(connectionId.Id, connection))
                         {
                             throw new HazelException("Failed to add a connection. This should never happen.");
                         }
@@ -297,10 +323,10 @@ namespace Hazel.Udp.FewerThreads
         /// <summary>
         ///     Removes a virtual connection from the list.
         /// </summary>
-        /// <param name="endPoint">The endpoint of the virtual connection.</param>
-        internal bool RemoveConnectionTo(EndPoint endPoint)
+        /// <param name="endPoint">Connection key of the virtual connection.</param>
+        internal bool RemoveConnectionTo(ConnectionId connectionId)
         {
-            return this.allConnections.TryRemove(endPoint, out var conn);
+            return this.allConnections.TryRemove(connectionId.Id, out var conn);
         }
 
         protected virtual void Dispose(bool disposing)

diff --git a/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs b/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs
index 84acf5f871f40da1c19a627391e5451a8398a53b..31c9ca2e5329a1130a8624877589f302b0bfa163 100644 (file)
--- a/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs
+++ b/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs
@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Net;
 
 namespace Hazel.Udp.FewerThreads
@@ -20,16 +20,19 @@ namespace Hazel.Udp.FewerThreads
         /// </remarks>
         public ThreadLimitedUdpConnectionListener Listener { get; private set; }
 
+        private ThreadLimitedUdpConnectionListener.ConnectionId ConnectionId;
+
         /// <summary>
         ///     Creates a UdpConnection for the virtual connection to the endpoint.
         /// </summary>
         /// <param name="listener">The listener that created this connection.</param>
         /// <param name="endPoint">The endpoint that we are connected to.</param>
         /// <param name="IPMode">The IPMode we are connected using.</param>
-        internal ThreadLimitedUdpServerConnection(ThreadLimitedUdpConnectionListener listener, IPEndPoint endPoint, IPMode IPMode)
+        internal ThreadLimitedUdpServerConnection(ThreadLimitedUdpConnectionListener listener, ThreadLimitedUdpConnectionListener.ConnectionId connectionId, IPEndPoint endPoint, IPMode IPMode)
             : base()
         {
             this.Listener = listener;
+            this.ConnectionId = connectionId;
             this.RemoteEndPoint = endPoint;
             this.EndPoint = endPoint;
             this.IPMode = IPMode;
@@ -69,7 +72,7 @@ namespace Hazel.Udp.FewerThreads
         /// </summary>
         protected override bool SendDisconnect(MessageWriter data = null)
         {
-            if (!Listener.RemoveConnectionTo(RemoteEndPoint)) return false;
+            if (!Listener.RemoveConnectionTo(this.ConnectionId)) return false;
             this._state = ConnectionState.NotConnected;
             
             var bytes = EmptyDisconnectBytes;