Refactor serializer locking and thread instances.

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -7,7 +7,6 @@ using System.Threading.Tasks;
 using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
-using MQTTnet.Internal;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.ManagedClient
@@ -15,8 +14,7 @@ namespace MQTTnet.Extensions.ManagedClient
     public class ManagedMqttClient : IManagedMqttClient
     {
         private readonly BlockingCollection<ManagedMqttApplicationMessage> _messageQueue = new BlockingCollection<ManagedMqttApplicationMessage>();
-        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
-        private readonly AsyncLock _subscriptionsLock = new AsyncLock();
+        private readonly ConcurrentDictionary<string, MqttQualityOfServiceLevel> _subscriptions = new ConcurrentDictionary<string, MqttQualityOfServiceLevel>();
         private readonly List<string> _unsubscriptions = new List<string>();
 
         private readonly IMqttClient _mqttClient;
@@ -118,39 +116,36 @@ namespace MQTTnet.Extensions.ManagedClient
             _messageQueue.Add(applicationMessage);
         }
 
-        public async Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
+        public Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
         {
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
 
-            using (await _subscriptionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
+            foreach (var topicFilter in topicFilters)
             {
-                foreach (var topicFilter in topicFilters)
+                _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                _subscriptionsNotPushed = true;
+            }
+
+            return Task.FromResult(0);
+        }
+
+        public Task UnsubscribeAsync(IEnumerable<string> topics)
+        {
+            foreach (var topic in topics)
+            {
+                if (_subscriptions.TryRemove(topic, out _))
                 {
-                    _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                    _unsubscriptions.Add(topic);
                     _subscriptionsNotPushed = true;
                 }
             }
-        }
 
-        public async Task UnsubscribeAsync(IEnumerable<string> topics)
-        {
-            using (await _subscriptionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
-            {
-                foreach (var topic in topics)
-                {
-                    if (_subscriptions.Remove(topic))
-                    {
-                        _unsubscriptions.Add(topic);
-                        _subscriptionsNotPushed = true;
-                    }
-                }
-            }
+            return Task.FromResult(0);
         }
 
         public void Dispose()
         {
             _messageQueue?.Dispose();
-            _subscriptionsLock?.Dispose();
             _connectionCancellationToken?.Dispose();
             _publishingCancellationToken?.Dispose();
         }
@@ -289,7 +284,7 @@ namespace MQTTnet.Extensions.ManagedClient
             List<TopicFilter> subscriptions;
             List<string> unsubscriptions;
 
-            using (await _subscriptionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
+            lock (_subscriptions)
             {
                 subscriptions = _subscriptions.Select(i => new TopicFilter(i.Key, i.Value)).ToList();
 

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -18,6 +18,8 @@ namespace MQTTnet.Adapter
         private const uint ErrorOperationAborted = 0x800703E3;
         private const int ReadBufferSize = 4096;  // TODO: Move buffer size to config
 
+        private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);
+
         private readonly IMqttNetChildLogger _logger;
         private readonly IMqttChannel _channel;
 
@@ -88,6 +90,7 @@ namespace MQTTnet.Adapter
 
         public async Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
         {
+            await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
             try
             {
                 _logger.Verbose("TX >>> {0}", packet);
@@ -107,6 +110,10 @@ namespace MQTTnet.Adapter
 
                 WrapException(exception);
             }
+            finally
+            {
+                _writerSemaphore.Release();
+            }
         }
 
         public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)

Source/MQTTnet/Diagnostics/MqttNetChildLogger.cs

@@ -9,7 +9,7 @@ namespace MQTTnet.Diagnostics
 
         public MqttNetChildLogger(IMqttNetLogger logger, string source)
         {
-            _logger = logger;
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
             _source = source;
         }
 

Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs

@@ -40,7 +40,18 @@ namespace MQTTnet.Implementations
 
         public static Func<MqttClientTcpOptions, IEnumerable<ChainValidationResult>> CustomIgnorableServerCertificateErrorsResolver { get; set; }
 
-        public string Endpoint => _socket?.Information?.RemoteAddress?.ToString(); // TODO: Check if contains also the port.
+        public string Endpoint
+        {
+            get
+            {
+                if (_socket?.Information != null)
+                {
+                    return _socket.Information.RemoteAddress + ":" + _socket.Information.RemotePort;
+                }
+
+                return null;
+            }
+        }
 
         public async Task ConnectAsync(CancellationToken cancellationToken)
         {
@@ -81,10 +92,13 @@ namespace MQTTnet.Implementations
             return _readStream.ReadAsync(buffer, offset, count, cancellationToken);
         }
 
-        public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
-            await _writeStream.WriteAsync(buffer, offset, count, cancellationToken);
-            await _writeStream.FlushAsync(cancellationToken);
+            // In the write method only the internal buffer will be filled. So here is no
+            // async/await required. The real network transmit is done when calling the
+            // Flush method.
+            _writeStream.Write(buffer, offset, count);
+            return _writeStream.FlushAsync(cancellationToken);
         }
 
         public void Dispose()

Source/MQTTnet/MQTTnet.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">netstandard1.3;netstandard2.0;net452;uap10.0</TargetFrameworks>
+    <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">netstandard1.3;netstandard2.0;net452;net461;uap10.0</TargetFrameworks>
     <TargetFrameworks Condition=" '$(OS)' != 'Windows_NT' ">netstandard1.3;netstandard2.0</TargetFrameworks>
     <AssemblyName>MQTTnet</AssemblyName>
     <RootNamespace>MQTTnet</RootNamespace>
@@ -62,4 +62,7 @@
   <ItemGroup Condition="'$(TargetFramework)'=='net452'">
   </ItemGroup>
 
+  <ItemGroup Condition="'$(TargetFramework)'=='net461'">
+  </ItemGroup>
+
 </Project>

Source/MQTTnet/Serializer/MqttPacketSerializer.cs

@@ -11,8 +11,7 @@ namespace MQTTnet.Serializer
     {
         private const int FixedHeaderSize = 1;
 
-        [ThreadStatic]
-        private static MqttPacketWriter _packetWriter;
+        private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter();
 
         public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
 
@@ -20,14 +19,12 @@ namespace MQTTnet.Serializer
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
-            var packetWriter = InitializePacketWriter();
-
             // Leave enough head space for max header size (fixed + 4 variable remaining length = 5 bytes)
-            packetWriter.Reset();
-            packetWriter.Seek(5);
+            _packetWriter.Reset();
+            _packetWriter.Seek(5);
 
-            var fixedHeader = SerializePacket(packet, packetWriter);
-            var remainingLength = packetWriter.Length - 5;
+            var fixedHeader = SerializePacket(packet, _packetWriter);
+            var remainingLength = _packetWriter.Length - 5;
 
             var remainingLengthBuffer = MqttPacketWriter.EncodeRemainingLength(remainingLength);
 
@@ -35,12 +32,12 @@ namespace MQTTnet.Serializer
             var headerOffset = 5 - headerSize;
 
             // Position cursor on correct offset on beginining of array (has leading 0x0)
-            packetWriter.Seek(headerOffset);
-            packetWriter.Write(fixedHeader);
-            packetWriter.Write(remainingLengthBuffer.Array, remainingLengthBuffer.Offset, remainingLengthBuffer.Count);
+            _packetWriter.Seek(headerOffset);
+            _packetWriter.Write(fixedHeader);
+            _packetWriter.Write(remainingLengthBuffer.Array, remainingLengthBuffer.Offset, remainingLengthBuffer.Count);
 
-            var buffer = packetWriter.GetBuffer();
-            return new ArraySegment<byte>(buffer, headerOffset, packetWriter.Length - headerOffset);
+            var buffer = _packetWriter.GetBuffer();
+            return new ArraySegment<byte>(buffer, headerOffset, _packetWriter.Length - headerOffset);
         }
 
         public MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket)
@@ -76,7 +73,7 @@ namespace MQTTnet.Serializer
 
         public void FreeBuffer()
         {
-            InitializePacketWriter().FreeBuffer();
+            _packetWriter.FreeBuffer();
         }
 
         private byte SerializePacket(MqttBasePacket packet, MqttPacketWriter packetWriter)
@@ -101,16 +98,6 @@ namespace MQTTnet.Serializer
             }
         }
 
-        private static MqttPacketWriter InitializePacketWriter()
-        {
-            if (_packetWriter == null)
-            {
-                _packetWriter = new MqttPacketWriter();
-            }
-
-            return _packetWriter;
-        }
-
         private static MqttBasePacket DeserializeUnsubAck(MqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);

Source/MQTTnet/Serializer/MqttPacketWriter.cs

@@ -12,6 +12,8 @@ namespace MQTTnet.Serializer
     /// </summary>
     public class MqttPacketWriter
     {
+        public static int MaxBufferSize = 4096;
+
         private byte[] _buffer = new byte[128];
 
         private int _position;
@@ -117,12 +119,12 @@ namespace MQTTnet.Serializer
             // a lot and the size will never reduced. So this method tries to find a size which can be held in 
             // memory for a long time without causing troubles.
 
-            if (_buffer.Length < 4096)
+            if (_buffer.Length < MaxBufferSize)
             {
                 return;
             }
 
-            Array.Resize(ref _buffer, 4096);
+            Array.Resize(ref _buffer, MaxBufferSize);
         }
 
         private void EnsureAdditionalCapacity(int additionalCapacity)

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -148,7 +148,7 @@ namespace MQTTnet.Server
                         clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage);
                     }
                 }
-                catch (TaskCanceledException)
+                catch (OperationCanceledException)
                 {
                 }
                 catch (Exception exception)