Merge pull request #220 from israellot/optmizations

Replace double dictionary with a tuple key

Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs

@@ -115,9 +115,13 @@ namespace MQTTnet.Adapter
                 ReceivedMqttPacket receivedMqttPacket = null;
                 try
                 {
+
                     if (timeout > TimeSpan.Zero)
                     {
-                        receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
+                        var timeoutCts = new CancellationTokenSource(timeout);
+                        var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCts.Token);
+
+                        receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, linkedCts.Token).ConfigureAwait(false);
                     }
                     else
                     {

Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs

@@ -10,7 +10,8 @@ namespace MQTTnet.Client
 {
     public class MqttPacketDispatcher
     {
-        private readonly ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>> _awaiters = new ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>>();
+        
+        private readonly ConcurrentDictionary<Tuple<ushort?, Type>, TaskCompletionSource<MqttBasePacket>> _awaiters = new ConcurrentDictionary<Tuple<ushort?, Type>, TaskCompletionSource<MqttBasePacket>>();
         private readonly IMqttNetLogger _logger;
 
         public MqttPacketDispatcher(IMqttNetLogger logger)
@@ -22,7 +23,7 @@ namespace MQTTnet.Client
         {
             var packetAwaiter = AddPacketAwaiter(responseType, identifier);
             try
-            {
+            {                
                 return await packetAwaiter.Task.TimeoutAfter(timeout).ConfigureAwait(false);
             }
             catch (MqttCommunicationTimedOutException)
@@ -40,21 +41,20 @@ namespace MQTTnet.Client
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
-            var type = packet.GetType();
+            ushort? identifier = 0;
-
+            if (packet is IMqttPacketWithIdentifier packetWithIdentifier)
-            if (_awaiters.TryGetValue(type, out var byId))
             {
-                ushort? identifier = 0;
+                identifier = packetWithIdentifier.PacketIdentifier;
-                if (packet is IMqttPacketWithIdentifier packetWithIdentifier)
+            }
-                {
-                    identifier = packetWithIdentifier.PacketIdentifier;
-                }
 
-                if (byId.TryRemove(identifier.Value, out var tcs))
+            var type = packet.GetType();
-                {
+            var key = new Tuple<ushort?, Type>(identifier, type);
-                    tcs.TrySetResult(packet);
+              
-                    return;
+
-                }
+            if (_awaiters.TryRemove(key, out var tcs))
+            {
+                tcs.TrySetResult(packet);
+                return;
             }
 
             throw new InvalidOperationException($"Packet of type '{type.Name}' not handled or dispatched.");
@@ -74,8 +74,8 @@ namespace MQTTnet.Client
                 identifier = 0;
             }
 
-            var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
+            var dictionaryKey = new Tuple<ushort?,Type>(identifier, responseType);            
-            if (!byId.TryAdd(identifier.Value, tcs))
+            if (!_awaiters.TryAdd(dictionaryKey,tcs))
             {
                 throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{responseType}' with identifier {identifier}.");
             }
@@ -90,8 +90,8 @@ namespace MQTTnet.Client
                 identifier = 0;
             }
 
-            var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
+            var dictionaryKey = new Tuple<ushort?, Type>(identifier, responseType);
-            byId.TryRemove(identifier.Value, out var _);
+            _awaiters.TryRemove(dictionaryKey, out var _);
         }
     }
 }

README.md

@@ -78,7 +78,7 @@ If you want to contribute to this project just create a pull request. But only p
 This library is used in the following projects:
 
 * MQTT Client Rx (Wrapper for Reactive Extensions, <https://github.com/1iveowl/MQTTClient.rx>)
-* MQTT Tester (MQTT client test app for [Android](https://play.google.com/store/apps/details?id=com.liveowl.mqtttester), [iOS](https://itunes.apple.com/us/app/mqtt-tester/id1278621826?mt=8), and [Windows 10](https://www.microsoft.com/en-us/store/p/mqtt-tester/9wzdncrd272c))
+* MQTT Tester (MQTT client test app for [Android](https://play.google.com/store/apps/details?id=com.liveowl.mqtttester) and [iOS](https://itunes.apple.com/us/app/mqtt-tester/id1278621826?mt=8))
 * Wirehome (Open Source Home Automation system for .NET, <https://github.com/chkr1011/Wirehome>)
 
 

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -75,7 +75,7 @@ namespace MQTTnet.Core.Tests
                 await s.StopAsync();
             }
 
-            Assert.AreEqual(1, receivedMessagesCount);
+            Assert.AreEqual(0, receivedMessagesCount);
         }
 
         [TestMethod]