diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs
index b184b0ab..7f024f79 100644
--- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs
+++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs
@@ -230,13 +230,13 @@ namespace MQTTnet.Adapter
                 return null;
             }
 
+            if (readFixedHeaderResult.ConnectionClosed)
+            {
+                return null;
+            }
+
             try
             {
-                if (readFixedHeaderResult.ConnectionClosed)
-                {
-                    return null;
-                }
-
                 IsReadingPacket = true;
 
                 var fixedHeader = readFixedHeaderResult.FixedHeader;
@@ -245,9 +245,11 @@ namespace MQTTnet.Adapter
                     return new ReceivedMqttPacket(fixedHeader.Flags, null, 2);
                 }
 
-                var body = new byte[fixedHeader.RemainingLength];
+                var bodyLength = fixedHeader.RemainingLength;
+                var body = new byte[bodyLength];
+
                 var bodyOffset = 0;
-                var chunkSize = Math.Min(ReadBufferSize, fixedHeader.RemainingLength);
+                var chunkSize = Math.Min(ReadBufferSize, bodyLength);
 
                 do
                 {
@@ -270,9 +272,9 @@ namespace MQTTnet.Adapter
                     }
 
                     bodyOffset += readBytes;
-                } while (bodyOffset < body.Length);
+                } while (bodyOffset < bodyLength);
 
-                var bodyReader = new MqttPacketBodyReader(body, 0, body.Length);
+                var bodyReader = new MqttPacketBodyReader(body, 0, bodyLength);
                 return new ReceivedMqttPacket(fixedHeader.Flags, bodyReader, fixedHeader.TotalLength);
             }
             finally
diff --git a/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs b/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs
index 355ad96f..783f1d82 100644
--- a/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs
+++ b/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs
@@ -2,7 +2,7 @@
 
 namespace MQTTnet.Adapter
 {
-    public class ReceivedMqttPacket
+    public sealed class ReceivedMqttPacket
     {
         public ReceivedMqttPacket(byte fixedHeader, IMqttPacketBodyReader body, int totalLength)
         {
@@ -11,7 +11,7 @@ namespace MQTTnet.Adapter
             TotalLength = totalLength;
         }
 
-        public byte FixedHeader { get; set; }
+        public byte FixedHeader { get; }
 
         public IMqttPacketBodyReader Body { get; }
 
diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs
index 0102be15..cbff63aa 100644
--- a/Source/MQTTnet/Client/MqttClient.cs
+++ b/Source/MQTTnet/Client/MqttClient.cs
@@ -217,8 +217,6 @@ namespace MQTTnet.Client
 
         public Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
         {
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
-
             MqttTopicValidator.ThrowIfInvalid(applicationMessage);
 
             ThrowIfDisposed();
diff --git a/Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs b/Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs
index 2e0f8f02..fdf8764e 100644
--- a/Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs
+++ b/Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs
@@ -1,6 +1,4 @@
-using System;
-
-namespace MQTTnet.Formatter
+namespace MQTTnet.Formatter
 {
     public interface IMqttPacketBodyReader
     {
diff --git a/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs b/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs
index ea0fa1a5..82478e12 100644
--- a/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs
+++ b/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs
@@ -6,13 +6,13 @@ using MQTTnet.Exceptions;
 
 namespace MQTTnet.Formatter
 {
-    public class MqttPacketBodyReader : IMqttPacketBodyReader
+    public sealed class MqttPacketBodyReader : IMqttPacketBodyReader
     {
-        private readonly byte[] _buffer;
-        private readonly int _initialOffset;
-        private readonly int _length;
+        readonly byte[] _buffer;
+        readonly int _initialOffset;
+        readonly int _length;
 
-        private int _offset;
+        int _offset;
 
         public MqttPacketBodyReader(byte[] buffer, int offset, int length)
         {
diff --git a/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs b/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs
index 9469427d..6059d36e 100644
--- a/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs
+++ b/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs
@@ -7,8 +7,12 @@ namespace MQTTnet.Implementations
     {
 #if NET452
         public static Task CompletedTask => Task.FromResult(0);
+
+        public static byte[] EmptyByteArray { get; } = new byte[0];
 #else
         public static Task CompletedTask => Task.CompletedTask;
+
+        public static byte[] EmptyByteArray { get; } = Array.Empty<byte>();
 #endif
 
         public static void Sleep(TimeSpan timeout)
diff --git a/Source/MQTTnet/Internal/AsyncLock.cs b/Source/MQTTnet/Internal/AsyncLock.cs
index b5b614f6..6e1cef35 100644
--- a/Source/MQTTnet/Internal/AsyncLock.cs
+++ b/Source/MQTTnet/Internal/AsyncLock.cs
@@ -16,28 +16,31 @@ namespace MQTTnet.Internal
             _releaser = Task.FromResult((IDisposable)new Releaser(this));
         }
 
-        public Task<IDisposable> WaitAsync()
-        {
-            return WaitAsync(CancellationToken.None);
-        }
-
         public Task<IDisposable> WaitAsync(CancellationToken cancellationToken)
         {
             Task task;
+
+            // This lock is required to avoid ObjectDisposedExceptions.
+            // These are fired when this lock gets disposed (and thus the semaphore)
+            // and a worker thread tries to call this method at the same time.
+            // Another way would be catching all ObjectDisposedExceptions but this situation happens
+            // quite often when clients are disconnecting.
             lock (_syncRoot)
             {
-                if (_semaphore == null)
-                {
-                    throw new ObjectDisposedException("The AsyncLock is disposed.");
-                }
-
-                task = _semaphore.WaitAsync(cancellationToken);
-                if (task.Status == TaskStatus.RanToCompletion)
-                {
-                    return _releaser;
-                }
+                task = _semaphore?.WaitAsync(cancellationToken);
             }
 
+            if (task == null)
+            {
+                throw new ObjectDisposedException("The AsyncLock is disposed.");
+            }
+
+            if (task.Status == TaskStatus.RanToCompletion)
+            {
+                return _releaser;
+            }
+
+            // Wait for the _WaitAsync_ method and return the releaser afterwards.
             return task.ContinueWith(
                 (_, state) => (IDisposable)state, 
                 _releaser.Result, 
diff --git a/Source/MQTTnet/Protocol/MqttTopicValidator.cs b/Source/MQTTnet/Protocol/MqttTopicValidator.cs
index 9787e95d..00b16793 100644
--- a/Source/MQTTnet/Protocol/MqttTopicValidator.cs
+++ b/Source/MQTTnet/Protocol/MqttTopicValidator.cs
@@ -9,12 +9,17 @@ namespace MQTTnet.Protocol
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
-            if (applicationMessage.TopicAlias > 0)
+            if (!applicationMessage.TopicAlias.HasValue)
             {
-                return;
+                ThrowIfInvalid(applicationMessage.Topic);
+            }
+            else
+            {
+                if (applicationMessage.TopicAlias.Value == 0)
+                {
+                    throw new MqttProtocolViolationException("The topic alias cannot be 0.");
+                }
             }
-
-            ThrowIfInvalid(applicationMessage.Topic);
         }
 
         public static void ThrowIfInvalid(string topic)
diff --git a/Source/MQTTnet/Server/CheckSubscriptionsResult.cs b/Source/MQTTnet/Server/CheckSubscriptionsResult.cs
index 9e2d1f0d..1d553807 100644
--- a/Source/MQTTnet/Server/CheckSubscriptionsResult.cs
+++ b/Source/MQTTnet/Server/CheckSubscriptionsResult.cs
@@ -2,7 +2,7 @@
 
 namespace MQTTnet.Server
 {
-    public class CheckSubscriptionsResult
+    public struct CheckSubscriptionsResult
     {
         public bool IsSubscribed { get; set; }
 
diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs
index 80c60210..811db7d1 100644
--- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs
+++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs
@@ -18,7 +18,7 @@ namespace MQTTnet.Server
 {
     public sealed class MqttClientSessionsManager : IDisposable
     {
-        readonly AsyncQueue<MqttEnqueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttEnqueuedApplicationMessage>();
+        readonly BlockingCollection<MqttEnqueuedApplicationMessage> _messageQueue = new BlockingCollection<MqttEnqueuedApplicationMessage>();
 
         readonly object _createConnectionSyncRoot = new object();
         readonly Dictionary<string, MqttClientConnection> _connections = new Dictionary<string, MqttClientConnection>();
@@ -176,7 +176,7 @@ namespace MQTTnet.Server
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
-            _messageQueue.Enqueue(new MqttEnqueuedApplicationMessage(applicationMessage, sender));
+            _messageQueue.Add(new MqttEnqueuedApplicationMessage(applicationMessage, sender));
         }
 
         public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
@@ -270,46 +270,41 @@ namespace MQTTnet.Server
         {
             try
             {
-                if (cancellationToken.IsCancellationRequested)
-                {
-                    return;
-                }
-
-                var dequeueResult = await _messageQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
-                if (!dequeueResult.IsSuccess)
-                {
-                    return;
-                }
-
-                var queuedApplicationMessage = dequeueResult.Item;
+                cancellationToken.ThrowIfCancellationRequested();
 
+                var queuedApplicationMessage = _messageQueue.Take(cancellationToken);
                 var sender = queuedApplicationMessage.Sender;
+                var senderClientId = sender?.ClientId ?? _options.ClientId;
                 var applicationMessage = queuedApplicationMessage.ApplicationMessage;
 
-                var interceptorContext = await InterceptApplicationMessageAsync(sender, applicationMessage).ConfigureAwait(false);
-                if (interceptorContext != null)
+                var interceptor = _options.ApplicationMessageInterceptor;
+                if (interceptor != null)
                 {
-                    if (interceptorContext.CloseConnection)
+                    var interceptorContext = await InterceptApplicationMessageAsync(interceptor, sender, applicationMessage).ConfigureAwait(false);
+                    if (interceptorContext != null)
                     {
-                        if (sender != null)
+                        if (interceptorContext.CloseConnection)
                         {
-                            await sender.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
+                            if (sender != null)
+                            {
+                                await sender.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
+                            }
                         }
-                    }
 
-                    if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
-                    {
-                        return;
-                    }
+                        if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
+                        {
+                            return;
+                        }
 
-                    applicationMessage = interceptorContext.ApplicationMessage;
+                        applicationMessage = interceptorContext.ApplicationMessage;
+                    }
                 }
-
-                await _eventDispatcher.SafeNotifyApplicationMessageReceivedAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false);
+                
+                await _eventDispatcher.SafeNotifyApplicationMessageReceivedAsync(senderClientId, applicationMessage).ConfigureAwait(false);
 
                 if (applicationMessage.Retain)
                 {
-                    await _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false);
+                    await _retainedMessagesManager.HandleMessageAsync(senderClientId, applicationMessage).ConfigureAwait(false);
                 }
 
                 var deliveryCount = 0;
@@ -318,11 +313,7 @@ namespace MQTTnet.Server
                 {
                     foreach (var clientSession in _sessions.Values)
                     {
-                        var isSubscribed = clientSession.EnqueueApplicationMessage(
-                            applicationMessage,
-                            sender?.ClientId,
-                            false);
-
+                        var isSubscribed = clientSession.EnqueueApplicationMessage(applicationMessage, senderClientId, false);
                         if (isSubscribed)
                         {
                             deliveryCount++;
@@ -333,13 +324,12 @@ namespace MQTTnet.Server
                 if (deliveryCount == 0)
                 {
                     var undeliveredMessageInterceptor = _options.UndeliveredMessageInterceptor;
-
                     if (undeliveredMessageInterceptor == null)
                     {
                         return;
                     }
 
-                    await undeliveredMessageInterceptor.InterceptApplicationMessagePublishAsync(new MqttApplicationMessageInterceptorContext(sender?.ClientId, sender?.Session?.Items, applicationMessage));
+                    await undeliveredMessageInterceptor.InterceptApplicationMessagePublishAsync(new MqttApplicationMessageInterceptorContext(senderClientId, sender?.Session?.Items, applicationMessage));
                 }
             }
             catch (OperationCanceledException)
@@ -423,14 +413,8 @@ namespace MQTTnet.Server
             }
         }
 
-        async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)
+        async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(IMqttServerApplicationMessageInterceptor interceptor, MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)
         {
-            var interceptor = _options.ApplicationMessageInterceptor;
-            if (interceptor == null)
-            {
-                return null;
-            }
-
             string senderClientId;
             IDictionary<object, object> sessionItems;
 
diff --git a/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs
index 2d216295..89d21f57 100644
--- a/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs
+++ b/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs
@@ -3,23 +3,26 @@ using MQTTnet.Protocol;
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace MQTTnet.Server
 {
     public sealed class MqttClientSubscriptionsManager
     {
+        // We use a reader writer lock because the subscriptions are only read most of the time.
+        // Since writing is done by multiple threads (server API or connection thread), we cannot avoid locking
+        // completely by swapping references etc.
+        readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim();
         readonly Dictionary<string, MqttTopicFilter> _subscriptions = new Dictionary<string, MqttTopicFilter>();
         readonly MqttClientSession _clientSession;
-        readonly IMqttServerOptions _serverOptions;
+        readonly IMqttServerOptions _options;
         readonly MqttServerEventDispatcher _eventDispatcher;
 
         public MqttClientSubscriptionsManager(MqttClientSession clientSession, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions)
         {
             _clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
-
-            // TODO: Consider removing the server options here and build a new class "ISubscriptionInterceptor" and just pass it. The instance is generated in the root server class upon start.
-            _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
+            _options = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
         }
 
@@ -52,15 +55,22 @@ namespace MQTTnet.Server
                     result.CloseConnection = true;
                 }
 
-                if (interceptorContext.AcceptSubscription && !string.IsNullOrEmpty(finalTopicFilter?.Topic))
+                if (!interceptorContext.AcceptSubscription || string.IsNullOrEmpty(finalTopicFilter?.Topic))
                 {
-                    lock (_subscriptions)
-                    {
-                        _subscriptions[finalTopicFilter.Topic] = finalTopicFilter;
-                    }
-
-                    await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false);
+                    continue;
                 }
+
+                _subscriptionsLock.EnterWriteLock();
+                try
+                {
+                    _subscriptions[finalTopicFilter.Topic] = finalTopicFilter;
+                }
+                finally
+                {
+                    _subscriptionsLock.ExitWriteLock();
+                }
+
+                await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false);
             }
 
             return result;
@@ -78,15 +88,22 @@ namespace MQTTnet.Server
                     continue;
                 }
 
-                if (interceptorContext.AcceptSubscription)
+                if (!interceptorContext.AcceptSubscription)
                 {
-                    lock (_subscriptions)
-                    {
-                        _subscriptions[topicFilter.Topic] = topicFilter;
-                    }
-
-                    await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
+                    continue;
                 }
+
+                _subscriptionsLock.EnterWriteLock();
+                try
+                {
+                    _subscriptions[topicFilter.Topic] = topicFilter;
+                }
+                finally
+                {
+                    _subscriptionsLock.ExitWriteLock();
+                }
+
+                await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
             }
         }
 
@@ -105,16 +122,16 @@ namespace MQTTnet.Server
                     continue;
                 }
 
-                lock (_subscriptions)
+                _subscriptionsLock.EnterWriteLock();
+                try
                 {
-                    if (_subscriptions.Remove(topicFilter))
-                    {
-                        reasonCodes.Add(MqttUnsubscribeReasonCode.Success);
-                    }
-                    else
-                    {
-                        reasonCodes.Add(MqttUnsubscribeReasonCode.NoSubscriptionExisted);
-                    }
+                    reasonCodes.Add(_subscriptions.Remove(topicFilter)
+                        ? MqttUnsubscribeReasonCode.Success
+                        : MqttUnsubscribeReasonCode.NoSubscriptionExisted);
+                }
+                finally
+                {
+                    _subscriptionsLock.ExitWriteLock();
                 }
             }
 
@@ -138,28 +155,40 @@ namespace MQTTnet.Server
                     continue;
                 }
 
-                lock (_subscriptions)
+                _subscriptionsLock.EnterWriteLock();
+                try
                 {
                     _subscriptions.Remove(topicFilter);
                 }
+                finally
+                {
+                    _subscriptionsLock.ExitWriteLock();
+                }
             }
         }
 
         public CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel qosLevel)
         {
-            var qosLevels = new HashSet<MqttQualityOfServiceLevel>();
-
-            lock (_subscriptions)
+            Dictionary<string, MqttTopicFilter> subscriptionsCopy;
+            _subscriptionsLock.EnterReadLock();
+            try
             {
-                foreach (var subscription in _subscriptions)
-                {
-                    if (!MqttTopicFilterComparer.IsMatch(topic, subscription.Key))
-                    {
-                        continue;
-                    }
+                subscriptionsCopy = new Dictionary<string, MqttTopicFilter>(_subscriptions);
+            }
+            finally
+            {
+                _subscriptionsLock.ExitReadLock();
+            }
 
-                    qosLevels.Add(subscription.Value.QualityOfServiceLevel);
+            var qosLevels = new HashSet<MqttQualityOfServiceLevel>();
+            foreach (var subscription in subscriptionsCopy)
+            {
+                if (!MqttTopicFilterComparer.IsMatch(topic, subscription.Key))
+                {
+                    continue;
                 }
+                
+                qosLevels.Add(subscription.Value.QualityOfServiceLevel);
             }
 
             if (qosLevels.Count == 0)
@@ -175,32 +204,20 @@ namespace MQTTnet.Server
 
         static MqttSubscribeReturnCode ConvertToSubscribeReturnCode(MqttQualityOfServiceLevel qualityOfServiceLevel)
         {
-            switch (qualityOfServiceLevel)
-            {
-                case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS0;
-                case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS1;
-                case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS2;
-                default: return MqttSubscribeReturnCode.Failure;
-            }
+            return (MqttSubscribeReturnCode)(int)qualityOfServiceLevel;
         }
 
         static MqttSubscribeReasonCode ConvertToSubscribeReasonCode(MqttQualityOfServiceLevel qualityOfServiceLevel)
         {
-            switch (qualityOfServiceLevel)
-            {
-                case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReasonCode.GrantedQoS0;
-                case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReasonCode.GrantedQoS1;
-                case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReasonCode.GrantedQoS2;
-                default: return MqttSubscribeReasonCode.UnspecifiedError;
-            }
+            return (MqttSubscribeReasonCode)(int)qualityOfServiceLevel;
         }
 
         async Task<MqttSubscriptionInterceptorContext> InterceptSubscribeAsync(MqttTopicFilter topicFilter)
         {
             var context = new MqttSubscriptionInterceptorContext(_clientSession.ClientId, topicFilter, _clientSession.Items);
-            if (_serverOptions.SubscriptionInterceptor != null)
+            if (_options.SubscriptionInterceptor != null)
             {
-                await _serverOptions.SubscriptionInterceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false);
+                await _options.SubscriptionInterceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false);
             }
 
             return context;
@@ -209,9 +226,9 @@ namespace MQTTnet.Server
         async Task<MqttUnsubscriptionInterceptorContext> InterceptUnsubscribeAsync(string topicFilter)
         {
             var context = new MqttUnsubscriptionInterceptorContext(_clientSession.ClientId, topicFilter, _clientSession.Items);
-            if (_serverOptions.UnsubscriptionInterceptor != null)
+            if (_options.UnsubscriptionInterceptor != null)
             {
-                await _serverOptions.UnsubscriptionInterceptor.InterceptUnsubscriptionAsync(context).ConfigureAwait(false);
+                await _options.UnsubscriptionInterceptor.InterceptUnsubscriptionAsync(context).ConfigureAwait(false);
             }
 
             return context;
diff --git a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
index 2d1f3b53..285f2246 100644
--- a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
+++ b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
@@ -1,17 +1,17 @@
 using MQTTnet.Diagnostics;
 using MQTTnet.Implementations;
-using MQTTnet.Internal;
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Internal;
 
 namespace MQTTnet.Server
 {
-    public class MqttRetainedMessagesManager : IMqttRetainedMessagesManager
+    public sealed class MqttRetainedMessagesManager : IMqttRetainedMessagesManager
     {
-        readonly byte[] _emptyArray = new byte[0];
-        readonly AsyncLock _messagesLock = new AsyncLock();
+        readonly AsyncLock _storageAccessLock = new AsyncLock();
         readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>();
 
         IMqttNetScopedLogger _logger;
@@ -37,12 +37,13 @@ namespace MQTTnet.Server
             try
             {
                 var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false);
-                if (retainedMessages?.Any() == true)
-                {
-                    using (await _messagesLock.WaitAsync().ConfigureAwait(false))
-                    {
-                        _messages.Clear();
 
+                lock (_messages)
+                {
+                    _messages.Clear();
+
+                    if (retainedMessages != null)
+                    {
                         foreach (var retainedMessage in retainedMessages)
                         {
                             _messages[retainedMessage.Topic] = retainedMessage;
@@ -62,9 +63,11 @@ namespace MQTTnet.Server
 
             try
             {
-                using (await _messagesLock.WaitAsync().ConfigureAwait(false))
+                List<MqttApplicationMessage> messagesForSave = null;
+                var saveIsRequired = false;
+                
+                lock (_messages)
                 {
-                    var saveIsRequired = false;
                     var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
 
                     if (!hasPayload)
@@ -81,7 +84,7 @@ namespace MQTTnet.Server
                         }
                         else
                         {
-                            if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? _emptyArray))
+                            if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? PlatformAbstractionLayer.EmptyByteArray))
                             {
                                 _messages[applicationMessage.Topic] = applicationMessage;
                                 saveIsRequired = true;
@@ -93,9 +96,16 @@ namespace MQTTnet.Server
 
                     if (saveIsRequired)
                     {
-                        if (_options.Storage != null)
+                        messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
+                    }
+                }
+
+                if (saveIsRequired)
+                {
+                    if (_options.Storage != null)
+                    {
+                        using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
                         {
-                            var messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
                             await _options.Storage.SaveRetainedMessagesAsync(messagesForSave).ConfigureAwait(false);
                         }
                     }
@@ -107,14 +117,14 @@ namespace MQTTnet.Server
             }
         }
 
-        public async Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<MqttTopicFilter> topicFilters)
+        public Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<MqttTopicFilter> topicFilters)
         {
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
 
             var matchingRetainedMessages = new List<MqttApplicationMessage>();
 
             List<MqttApplicationMessage> retainedMessages;
-            using (await _messagesLock.WaitAsync().ConfigureAwait(false))
+            lock (_messages)
             {
                 retainedMessages = _messages.Values.ToList();
             }
@@ -133,24 +143,27 @@ namespace MQTTnet.Server
                 }
             }
 
-            return matchingRetainedMessages;
+            return Task.FromResult((IList<MqttApplicationMessage>)matchingRetainedMessages);
         }
 
-        public async Task<IList<MqttApplicationMessage>> GetMessagesAsync()
+        public Task<IList<MqttApplicationMessage>> GetMessagesAsync()
         {
-            using (await _messagesLock.WaitAsync().ConfigureAwait(false))
+            lock (_messages)
             {
-                return _messages.Values.ToList();
+                return Task.FromResult((IList<MqttApplicationMessage>)_messages.Values.ToList());
             }
         }
 
         public async Task ClearMessagesAsync()
         {
-            using (await _messagesLock.WaitAsync().ConfigureAwait(false))
+            lock (_messages)
             {
                 _messages.Clear();
+            }
 
-                if (_options.Storage != null)
+            if (_options.Storage != null)
+            {
+                using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
                 {
                     await _options.Storage.SaveRetainedMessagesAsync(new List<MqttApplicationMessage>()).ConfigureAwait(false);
                 }
diff --git a/Source/MQTTnet/Server/MqttServerEventDispatcher.cs b/Source/MQTTnet/Server/MqttServerEventDispatcher.cs
index da413838..8a416a78 100644
--- a/Source/MQTTnet/Server/MqttServerEventDispatcher.cs
+++ b/Source/MQTTnet/Server/MqttServerEventDispatcher.cs
@@ -108,7 +108,7 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)).ConfigureAwait(false); ;
+                await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
diff --git a/Source/MQTTnet/Server/MqttTopicFilterComparer.cs b/Source/MQTTnet/Server/MqttTopicFilterComparer.cs
index 0b0fe6b2..4ba614d4 100644
--- a/Source/MQTTnet/Server/MqttTopicFilterComparer.cs
+++ b/Source/MQTTnet/Server/MqttTopicFilterComparer.cs
@@ -4,14 +4,14 @@ namespace MQTTnet.Server
 {
     public static class MqttTopicFilterComparer
     {
-        private const char LevelSeparator = '/';
-        private const char MultiLevelWildcard = '#';
-        private const char SingleLevelWildcard = '+';
+        const char LevelSeparator = '/';
+        const char MultiLevelWildcard = '#';
+        const char SingleLevelWildcard = '+';
 
         public static bool IsMatch(string topic, string filter)
         {
-            if (string.IsNullOrEmpty(topic)) throw new ArgumentNullException(nameof(topic));
-            if (string.IsNullOrEmpty(filter)) throw new ArgumentNullException(nameof(filter));
+            if (topic == null) throw new ArgumentNullException(nameof(topic));
+            if (filter == null) throw new ArgumentNullException(nameof(filter));
 
             var sPos = 0;
             var sLen = filter.Length;
diff --git a/Tests/MQTTnet.Core.Tests/AsyncLock_Tests.cs b/Tests/MQTTnet.Core.Tests/AsyncLock_Tests.cs
index a4e66702..dfdce18f 100644
--- a/Tests/MQTTnet.Core.Tests/AsyncLock_Tests.cs
+++ b/Tests/MQTTnet.Core.Tests/AsyncLock_Tests.cs
@@ -17,7 +17,7 @@ namespace MQTTnet.Tests
             var @lock = new AsyncLock();
             for (var i = 0; i < 100; i++)
             {
-                using (await @lock.WaitAsync().ConfigureAwait(false))
+                using (await @lock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
                 {
                     sum++;
                 }
@@ -33,7 +33,7 @@ namespace MQTTnet.Tests
             var @lock = new AsyncLock();
 
             // This call will never "release" the lock due to missing _using_.
-            await @lock.WaitAsync().ConfigureAwait(false);
+            await @lock.WaitAsync(CancellationToken.None).ConfigureAwait(false);
 
             using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3)))
             {
diff --git a/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj b/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj
index 4fffc337..3c4c8e88 100644
--- a/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj
+++ b/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj
@@ -21,4 +21,10 @@
     <ProjectReference Include="..\..\Source\MQTTnet\MQTTnet.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <None Update="Start.bat">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
 </Project>
diff --git a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs
index 63baf758..60038ef8 100644
--- a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs
+++ b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs
@@ -67,8 +67,7 @@ namespace MQTTnet.TestApp.NetCore
                     {
                         Server = "127.0.0.1"
                     },
-                    CleanSession = true,
-                    //KeepAlivePeriod = TimeSpan.FromSeconds(1)
+                    CleanSession = true
                 };
 
                 var client = new MqttFactory().CreateMqttClient();
diff --git a/Tests/MQTTnet.TestApp.NetCore/Start.bat b/Tests/MQTTnet.TestApp.NetCore/Start.bat
new file mode 100644
index 00000000..7ef0d879
--- /dev/null
+++ b/Tests/MQTTnet.TestApp.NetCore/Start.bat
@@ -0,0 +1 @@
+dotnet MQTTnet.TestApp.NetCore.dll
\ No newline at end of file