Merge pull request #1 from chkr1011/develop

Develop

Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs

@@ -0,0 +1,34 @@
+using System;
+using System.Net;
+using MQTTnet.Adapter;
+using MQTTnet.AspNetCore.Client.Tcp;
+using MQTTnet.Client;
+using MQTTnet.Diagnostics;
+using MQTTnet.Serializer;
+
+namespace MQTTnet.AspNetCore.Client
+{
+    public class MqttClientConnectionContextFactory : IMqttClientAdapterFactory
+    {
+        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+
+            var serializer = new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion };
+
+            switch (options.ChannelOptions)
+            {
+                case MqttClientTcpOptions tcpOptions:
+                    {
+                        var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort());
+                        var tcpConnection = new TcpConnection(endpoint);
+                        return new MqttConnectionContext(serializer, tcpConnection);
+                    }
+                default:
+                    {
+                        throw new NotSupportedException();
+                    }
+            }
+        }
+    }
+}

Source/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs

@@ -0,0 +1,22 @@
+using System;
+using System.Runtime.InteropServices;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+    public static class BufferExtensions
+    {
+        public static ArraySegment<byte> GetArray(this Memory<byte> memory)
+        {
+            return ((ReadOnlyMemory<byte>)memory).GetArray();
+        }
+
+        public static ArraySegment<byte> GetArray(this ReadOnlyMemory<byte> memory)
+        {
+            if (!MemoryMarshal.TryGetArray(memory, out var result))
+            {
+                throw new InvalidOperationException("Buffer backed by array was expected");
+            }
+            return result;
+        }
+    }
+}

Source/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs

@@ -0,0 +1,41 @@
+using System.IO.Pipelines;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+    public class DuplexPipe : IDuplexPipe
+    {
+        public DuplexPipe(PipeReader reader, PipeWriter writer)
+        {
+            Input = reader;
+            Output = writer;
+        }
+
+        public PipeReader Input { get; }
+
+        public PipeWriter Output { get; }
+
+        public static DuplexPipePair CreateConnectionPair(PipeOptions inputOptions, PipeOptions outputOptions)
+        {
+            var input = new Pipe(inputOptions);
+            var output = new Pipe(outputOptions);
+
+            var transportToApplication = new DuplexPipe(output.Reader, input.Writer);
+            var applicationToTransport = new DuplexPipe(input.Reader, output.Writer);
+
+            return new DuplexPipePair(applicationToTransport, transportToApplication);
+        }
+
+        // This class exists to work around issues with value tuple on .NET Framework
+        public readonly struct DuplexPipePair
+        {
+            public IDuplexPipe Transport { get; }
+            public IDuplexPipe Application { get; }
+
+            public DuplexPipePair(IDuplexPipe transport, IDuplexPipe application)
+            {
+                Transport = transport;
+                Application = application;
+            }
+        }
+    }
+}

Source/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs

@@ -0,0 +1,70 @@
+using System;
+using System.Diagnostics;
+using System.IO.Pipelines;
+using System.Net.Sockets;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+    public class SocketAwaitable : ICriticalNotifyCompletion
+    {
+        private static readonly Action _callbackCompleted = () => { };
+
+        private readonly PipeScheduler _ioScheduler;
+
+        private Action _callback;
+        private int _bytesTransferred;
+        private SocketError _error;
+
+        public SocketAwaitable(PipeScheduler ioScheduler)
+        {
+            _ioScheduler = ioScheduler;
+        }
+
+        public bool IsCompleted => ReferenceEquals(_callback, _callbackCompleted);
+
+        public SocketAwaitable GetAwaiter() => this;
+
+        public int GetResult()
+        {
+            Debug.Assert(ReferenceEquals(_callback, _callbackCompleted));
+
+            _callback = null;
+
+            if (_error != SocketError.Success)
+            {
+                throw new SocketException((int)_error);
+            }
+
+            return _bytesTransferred;
+        }
+
+        public void OnCompleted(Action continuation)
+        {
+            if (ReferenceEquals(_callback, _callbackCompleted) ||
+                ReferenceEquals(Interlocked.CompareExchange(ref _callback, continuation, null), _callbackCompleted))
+            {
+                Task.Run(continuation);
+            }
+        }
+
+        public void UnsafeOnCompleted(Action continuation)
+        {
+            OnCompleted(continuation);
+        }
+
+        public void Complete(int bytesTransferred, SocketError socketError)
+        {
+            _error = socketError;
+            _bytesTransferred = bytesTransferred;
+            var continuation = Interlocked.Exchange(ref _callback, _callbackCompleted);
+
+            if (continuation != null)
+            {
+                _ioScheduler.Schedule(state => ((Action)state)(), continuation);
+            }
+        }
+    }
+}

Source/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs

@@ -0,0 +1,37 @@
+using System;
+using System.IO.Pipelines;
+using System.Net.Sockets;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+    public class SocketReceiver
+    {
+        private readonly Socket _socket;
+        private readonly SocketAsyncEventArgs _eventArgs = new SocketAsyncEventArgs();
+        private readonly SocketAwaitable _awaitable;
+
+        public SocketReceiver(Socket socket, PipeScheduler scheduler)
+        {
+            _socket = socket;
+            _awaitable = new SocketAwaitable(scheduler);
+            _eventArgs.UserToken = _awaitable;
+            _eventArgs.Completed += (_, e) => ((SocketAwaitable)e.UserToken).Complete(e.BytesTransferred, e.SocketError);
+        }
+
+        public SocketAwaitable ReceiveAsync(Memory<byte> buffer)
+        {
+#if NETCOREAPP2_1
+            _eventArgs.SetBuffer(buffer);
+#else
+            var segment = buffer.GetArray();
+            _eventArgs.SetBuffer(segment.Array, segment.Offset, segment.Count);
+#endif
+            if (!_socket.ReceiveAsync(_eventArgs))
+            {
+                _awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError);
+            }
+
+            return _awaitable;
+        }
+    }
+}

Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs

@@ -0,0 +1,97 @@
+using System;
+using System.Buffers;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO.Pipelines;
+using System.Net.Sockets;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+    public class SocketSender
+    {
+        private readonly Socket _socket;
+        private readonly SocketAsyncEventArgs _eventArgs = new SocketAsyncEventArgs();
+        private readonly SocketAwaitable _awaitable;
+
+        private List<ArraySegment<byte>> _bufferList;
+
+        public SocketSender(Socket socket, PipeScheduler scheduler)
+        {
+            _socket = socket;
+            _awaitable = new SocketAwaitable(scheduler);
+            _eventArgs.UserToken = _awaitable;
+            _eventArgs.Completed += (_, e) => ((SocketAwaitable)e.UserToken).Complete(e.BytesTransferred, e.SocketError);
+        }
+
+        public SocketAwaitable SendAsync(in ReadOnlySequence<byte> buffers)
+        {
+            if (buffers.IsSingleSegment)
+            {
+                return SendAsync(buffers.First);
+            }
+
+#if NETCOREAPP2_1
+            if (!_eventArgs.MemoryBuffer.Equals(Memory<byte>.Empty))
+#else
+            if (_eventArgs.Buffer != null)
+#endif
+            {
+                _eventArgs.SetBuffer(null, 0, 0);
+            }
+
+            _eventArgs.BufferList = GetBufferList(buffers);
+
+            if (!_socket.SendAsync(_eventArgs))
+            {
+                _awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError);
+            }
+
+            return _awaitable;
+        }
+
+        private SocketAwaitable SendAsync(ReadOnlyMemory<byte> memory)
+        {
+            // The BufferList getter is much less expensive then the setter.
+            if (_eventArgs.BufferList != null)
+            {
+                _eventArgs.BufferList = null;
+            }
+
+#if NETCOREAPP2_1
+            _eventArgs.SetBuffer(MemoryMarshal.AsMemory(memory));
+#else
+            var segment = memory.GetArray();
+            _eventArgs.SetBuffer(segment.Array, segment.Offset, segment.Count);
+#endif
+            if (!_socket.SendAsync(_eventArgs))
+            {
+                _awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError);
+            }
+
+            return _awaitable;
+        }
+
+        private List<ArraySegment<byte>> GetBufferList(in ReadOnlySequence<byte> buffer)
+        {
+            Debug.Assert(!buffer.IsEmpty);
+            Debug.Assert(!buffer.IsSingleSegment);
+
+            if (_bufferList == null)
+            {
+                _bufferList = new List<ArraySegment<byte>>();
+            }
+            else
+            {
+                // Buffers are pooled, so it's OK to root them until the next multi-buffer write.
+                _bufferList.Clear();
+            }
+
+            foreach (var b in buffer)
+            {
+                _bufferList.Add(b.GetArray());
+            }
+
+            return _bufferList;
+        }
+    }
+}

Source/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs

@@ -0,0 +1,258 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.IO.Pipelines;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.Http.Features;
+using MQTTnet.Exceptions;
+
+namespace MQTTnet.AspNetCore.Client.Tcp
+{
+    public class TcpConnection : ConnectionContext
+    {
+        private volatile bool _aborted;
+        private readonly EndPoint _endPoint;
+        private SocketSender _sender;
+        private SocketReceiver _receiver;
+
+        private Socket _socket;
+        private IDuplexPipe _application;
+
+        public bool IsConnected { get; private set; }
+        public override string ConnectionId { get; set; }
+        public override IFeatureCollection Features { get; }
+        public override IDictionary<object, object> Items { get; set; }
+        public override IDuplexPipe Transport { get; set; }
+
+        public TcpConnection(EndPoint endPoint)
+        {
+            _endPoint = endPoint;
+        }
+
+        public TcpConnection(Socket socket)
+        {
+            _socket = socket;
+            _endPoint = socket.RemoteEndPoint;
+
+            _sender = new SocketSender(_socket, PipeScheduler.ThreadPool);
+            _receiver = new SocketReceiver(_socket, PipeScheduler.ThreadPool);
+        }
+
+        public Task DisposeAsync()
+        {
+            IsConnected = false;
+
+            Transport?.Output.Complete();
+            Transport?.Input.Complete();
+
+            _socket?.Dispose();
+
+            return Task.CompletedTask;
+        }
+
+        public async Task StartAsync()
+        {
+            if (_socket == null)
+            {
+                _socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+                _sender = new SocketSender(_socket, PipeScheduler.ThreadPool);
+                _receiver = new SocketReceiver(_socket, PipeScheduler.ThreadPool);
+                await _socket.ConnectAsync(_endPoint);
+            }
+
+            var pair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);
+
+            Transport = pair.Transport;
+            _application = pair.Application;
+
+            _ = ExecuteAsync();
+
+            IsConnected = true;
+        }
+
+        private async Task ExecuteAsync()
+        {
+            Exception sendError = null;
+            try
+            {
+                // Spawn send and receive logic
+                var receiveTask = DoReceive();
+                var sendTask = DoSend();
+
+                // If the sending task completes then close the receive
+                // We don't need to do this in the other direction because the kestrel
+                // will trigger the output closing once the input is complete.
+                if (await Task.WhenAny(receiveTask, sendTask) == sendTask)
+                {
+                    // Tell the reader it's being aborted
+                    _socket.Dispose();
+                }
+
+                // Now wait for both to complete
+                await receiveTask;
+                sendError = await sendTask;
+
+                // Dispose the socket(should noop if already called)
+                _socket.Dispose();
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine($"Unexpected exception in {nameof(TcpConnection)}.{nameof(StartAsync)}: " + ex);
+            }
+            finally
+            {
+                // Complete the output after disposing the socket
+                _application.Input.Complete(sendError);
+            }
+        }
+        private async Task DoReceive()
+        {
+            Exception error = null;
+
+            try
+            {
+                await ProcessReceives();
+            }
+            catch (SocketException ex) when (ex.SocketErrorCode == SocketError.ConnectionReset)
+            {
+                error = new MqttCommunicationException(ex);
+            }
+            catch (SocketException ex) when (ex.SocketErrorCode == SocketError.OperationAborted ||
+                                             ex.SocketErrorCode == SocketError.ConnectionAborted ||
+                                             ex.SocketErrorCode == SocketError.Interrupted ||
+                                             ex.SocketErrorCode == SocketError.InvalidArgument)
+            {
+                if (!_aborted)
+                {
+                    // Calling Dispose after ReceiveAsync can cause an "InvalidArgument" error on *nix.
+                    error = ConnectionAborted();
+                }
+            }
+            catch (ObjectDisposedException)
+            {
+                if (!_aborted)
+                {
+                    error = ConnectionAborted();
+                }
+            }
+            catch (IOException ex)
+            {
+                error = ex;
+            }
+            catch (Exception ex)
+            {
+                error = new IOException(ex.Message, ex);
+            }
+            finally
+            {
+                if (_aborted)
+                {
+                    error = error ?? ConnectionAborted();
+                }
+
+                _application.Output.Complete(error);
+            }
+        }
+
+        private async Task ProcessReceives()
+        {
+            while (true)
+            {
+                // Ensure we have some reasonable amount of buffer space
+                var buffer = _application.Output.GetMemory();
+
+                var bytesReceived = await _receiver.ReceiveAsync(buffer);
+
+                if (bytesReceived == 0)
+                {
+                    // FIN
+                    break;
+                }
+
+                _application.Output.Advance(bytesReceived);
+
+                var flushTask = _application.Output.FlushAsync();
+
+                if (!flushTask.IsCompleted)
+                {
+                    await flushTask;
+                }
+
+                var result = flushTask.GetAwaiter().GetResult();
+                if (result.IsCompleted)
+                {
+                    // Pipe consumer is shut down, do we stop writing
+                    break;
+                }
+            }
+        }
+
+        private Exception ConnectionAborted()
+        {
+            return new MqttCommunicationException("Connection Aborted");
+        }
+
+        private async Task<Exception> DoSend()
+        {
+            Exception error = null;
+
+            try
+            {
+                await ProcessSends();
+            }
+            catch (SocketException ex) when (ex.SocketErrorCode == SocketError.OperationAborted)
+            {
+            }
+            catch (ObjectDisposedException)
+            {
+            }
+            catch (IOException ex)
+            {
+                error = ex;
+            }
+            catch (Exception ex)
+            {
+                error = new IOException(ex.Message, ex);
+            }
+            finally
+            {
+                _aborted = true;
+                _socket.Shutdown(SocketShutdown.Both);
+            }
+
+            return error;
+        }
+
+        private async Task ProcessSends()
+        {
+            while (true)
+            {
+                // Wait for data to write from the pipe producer
+                var result = await _application.Input.ReadAsync();
+                var buffer = result.Buffer;
+
+                if (result.IsCanceled)
+                {
+                    break;
+                }
+
+                var end = buffer.End;
+                var isCompleted = result.IsCompleted;
+                if (!buffer.IsEmpty)
+                {
+                    await _sender.SendAsync(buffer);
+                }
+
+                _application.Input.AdvanceTo(end);
+
+                if (isCompleted)
+                {
+                    break;
+                }
+            }
+        }
+    }
+}

Source/MQTTnet.AspnetCore/ConnectionBuilderExtensions.cs

@@ -0,0 +1,12 @@
+using Microsoft.AspNetCore.Connections;
+
+namespace MQTTnet.AspNetCore
+{
+    public static class ConnectionBuilderExtensions
+    {
+        public static IConnectionBuilder UseMqtt(this IConnectionBuilder builder)
+        {
+            return builder.UseConnectionHandler<MqttConnectionHandler>();
+        }
+    }
+}

Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj

@@ -9,6 +9,7 @@
     <Company />
     <Authors />
     <PackageId />
+    <LangVersion>7.2</LangVersion>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">
@@ -16,7 +17,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.0.1" />
+    <PackageReference Include="Microsoft.AspNetCore.Connections.Abstractions" Version="2.1.0" />
     <PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.0.1" />
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.0.1" />
   </ItemGroup>

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -0,0 +1,117 @@
+using Microsoft.AspNetCore.Connections;
+using MQTTnet.Adapter;
+using MQTTnet.AspNetCore.Client.Tcp;
+using MQTTnet.Packets;
+using MQTTnet.Serializer;
+using System;
+using System.IO.Pipelines;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.AspNetCore
+{
+    public class MqttConnectionContext : IMqttChannelAdapter
+    {
+        public MqttConnectionContext(
+            IMqttPacketSerializer packetSerializer,
+            ConnectionContext connection)
+        {
+            PacketSerializer = packetSerializer;
+            Connection = connection;
+        }
+
+        public string Endpoint => Connection.ConnectionId;
+        public ConnectionContext Connection { get; }
+        public IMqttPacketSerializer PacketSerializer { get; }
+        public event EventHandler ReadingPacketStarted;
+        public event EventHandler ReadingPacketCompleted;
+
+        public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        {
+            if (Connection is TcpConnection tcp && !tcp.IsConnected)
+            {
+                return tcp.StartAsync();
+            }
+            return Task.CompletedTask;
+        }
+
+        public Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        {
+            Connection.Transport.Input.Complete();
+            Connection.Transport.Output.Complete();
+
+            return Task.CompletedTask;
+        }
+
+        public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        {
+            var input = Connection.Transport.Input;
+
+            try
+            {
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    ReadResult readResult;
+                    var readTask = input.ReadAsync(cancellationToken);
+                    if (readTask.IsCompleted)
+                    {
+                        readResult = readTask.Result;
+                    }
+                    else
+                    {
+                        readResult = await readTask;
+                    }
+
+                    var buffer = readResult.Buffer;
+
+                    var consumed = buffer.Start;
+                    var observed = buffer.Start;
+
+                    try
+                    {
+                        if (!buffer.IsEmpty)
+                        {
+                            if (PacketSerializer.TryDeserialize(buffer, out var packet, out consumed, out observed))
+                            {
+                                return packet;
+                            }
+                            else
+                            {
+                                // we did receive something but the message is not yet complete
+                                ReadingPacketStarted?.Invoke(this, EventArgs.Empty);
+                            }
+                        }
+                        else if (readResult.IsCompleted)
+                        {
+                            break;
+                        }
+                    }
+                    finally
+                    {
+                        // The buffer was sliced up to where it was consumed, so we can just advance to the start.
+                        // We mark examined as buffer.End so that if we didn't receive a full frame, we'll wait for more data
+                        // before yielding the read again.
+                        input.AdvanceTo(consumed, observed);
+                    }
+                }
+            }
+            finally
+            {
+                ReadingPacketCompleted?.Invoke(this, EventArgs.Empty);
+            }
+
+            cancellationToken.ThrowIfCancellationRequested();
+            return null;
+        }
+
+        public Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
+        {
+            var buffer = PacketSerializer.Serialize(packet);
+            return Connection.Transport.Output.WriteAsync(buffer.AsMemory(), cancellationToken).AsTask();
+        }
+
+        public void Dispose()
+        {
+        }
+    }
+}

Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs

@@ -0,0 +1,40 @@
+using Microsoft.AspNetCore.Connections;
+using MQTTnet.Adapter;
+using MQTTnet.Serializer;
+using MQTTnet.Server;
+using System;
+using System.Threading.Tasks;
+
+namespace MQTTnet.AspNetCore
+{
+    public class MqttConnectionHandler : ConnectionHandler, IMqttServerAdapter
+    {        
+        public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
+
+        public override async Task OnConnectedAsync(ConnectionContext connection)
+        {
+            var serializer = new MqttPacketSerializer();
+            using (var adapter = new MqttConnectionContext(serializer, connection))
+            {
+                var args = new MqttServerAdapterClientAcceptedEventArgs(adapter);
+                ClientAccepted?.Invoke(this, args);
+
+                await args.SessionTask;
+            }
+        }
+
+        public Task StartAsync(IMqttServerOptions options)
+        {
+            return Task.CompletedTask;
+        }
+
+        public Task StopAsync()
+        {
+            return Task.CompletedTask;
+        }
+        
+        public void Dispose()
+        {
+        }
+    }
+}

Source/MQTTnet.AspnetCore/MqttHostedServer.cs

@@ -13,7 +13,8 @@ namespace MQTTnet.AspNetCore
     {
         private readonly IMqttServerOptions _options;
 
-        public MqttHostedServer(IMqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(adapters, logger.CreateChildLogger(nameof(MqttHostedServer)))
+        public MqttHostedServer(IMqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) 
+            : base(adapters, logger.CreateChildLogger(nameof(MqttHostedServer)))
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
         }

Source/MQTTnet.AspnetCore/ReaderExtensions.cs

@@ -0,0 +1,82 @@
+using System;
+using System.Buffers;
+using MQTTnet.Adapter;
+using MQTTnet.Exceptions;
+using MQTTnet.Packets;
+using MQTTnet.Serializer;
+
+namespace MQTTnet.AspNetCore
+{
+    public static class ReaderExtensions
+    {
+        public static bool TryDeserialize(this IMqttPacketSerializer serializer, in ReadOnlySequence<byte> input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed)
+        {
+            packet = null;
+            consumed = input.Start;
+            observed = input.End;
+            var copy = input;
+            if (copy.Length < 2)
+            {
+                return false;
+            }
+
+            var fixedheader = copy.First.Span[0];
+            if (!TryReadBodyLength(ref copy, out var bodyLength))
+            {
+                return false;
+            }
+
+            var bodySlice = copy.Slice(0, bodyLength);
+            packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(bodySlice.GetArray(), 0)));
+            consumed = bodySlice.End;
+            observed = bodySlice.End;
+            return true;
+        }
+
+        private static byte[] GetArray(this in ReadOnlySequence<byte> input)
+        {
+            if (input.IsSingleSegment)
+            {
+                return input.First.Span.ToArray();
+            }
+
+            // Should be rare
+            return input.ToArray();
+        }
+
+        private static bool TryReadBodyLength(ref ReadOnlySequence<byte> input, out int result)
+        {
+            // Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html.
+            var multiplier = 1;
+            var value = 0;
+            byte encodedByte;
+            var index = 1;
+            result = 0;
+
+            var temp = input.Slice(0, Math.Min(5, input.Length)).GetArray();
+
+            do
+            {
+                if (index == temp.Length)
+                {
+                    return false;
+                }
+                encodedByte = temp[index];
+                index++;
+
+                value += (byte)(encodedByte & 127) * multiplier;
+                if (multiplier > 128 * 128 * 128)
+                {
+                    throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", temp.AsSpan(1, index).ToArray())}).");
+                }
+
+                multiplier *= 128;
+            } while ((encodedByte & 128) != 0);
+
+            input = input.Slice(index);
+
+            result = value;
+            return true;
+        }
+    }
+}

Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs

@@ -30,10 +30,19 @@ namespace MQTTnet.AspNetCore
 
             if (options.DefaultEndpointOptions.IsEnabled)
             {
+                services.AddSingleton<MqttTcpServerAdapter>();
                 services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttTcpServerAdapter>());
             }
 
             return services;
         }
+
+        public static IServiceCollection AddMqttConnectionHandler(this IServiceCollection services)
+        {
+            services.AddSingleton<MqttConnectionHandler>();
+            services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttConnectionHandler>());
+
+            return services;
+        }
     }
 }

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.Extensions.Rpc/MqttRpcClient.cs

@@ -82,11 +82,11 @@ namespace MQTTnet.Extensions.Rpc
                         timeoutCts.Cancel(false);
                         return result;
                     }
-                    catch (TaskCanceledException taskCanceledException)
+                    catch (OperationCanceledException exception)
                     {
                         if (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
                         {
-                            throw new MqttCommunicationTimedOutException(taskCanceledException);
+                            throw new MqttCommunicationTimedOutException(exception);
                         }
                         else
                         {

Source/MQTTnet/Adapter/IMqttChannelAdapter.cs

@@ -20,7 +20,7 @@ namespace MQTTnet.Adapter
 
         Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken);
 
-        Task SendPacketAsync(TimeSpan timeout, MqttBasePacket packet, CancellationToken cancellationToken);
+        Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken);
 
         Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken);
     }

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;
 
@@ -40,52 +42,91 @@ namespace MQTTnet.Adapter
         public event EventHandler ReadingPacketStarted;
         public event EventHandler ReadingPacketCompleted;
 
-        public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
-            _logger.Verbose("Connecting [Timeout={0}]", timeout);
 
-            return ExecuteAndWrapExceptionAsync(() =>
-                Internal.TaskExtensions.TimeoutAfter(ct => _channel.ConnectAsync(ct), timeout, cancellationToken));
-        }
-
-        public Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
-        {
-            ThrowIfDisposed();
-            _logger.Verbose("Disconnecting [Timeout={0}]", timeout);
-
-            return ExecuteAndWrapExceptionAsync(() =>
-                Internal.TaskExtensions.TimeoutAfter(ct => _channel.DisconnectAsync(), timeout, cancellationToken));
-        }
-
-        public Task SendPacketAsync(TimeSpan timeout, MqttBasePacket packet, CancellationToken cancellationToken)
-        {
-            return ExecuteAndWrapExceptionAsync(() =>
+            try
             {
-                _logger.Verbose("TX >>> {0} [Timeout={1}]", packet, timeout);
+                _logger.Verbose("Connecting [Timeout={0}]", timeout);
+
+                await Internal.TaskExtensions
+                    .TimeoutAfterAsync(ct => _channel.ConnectAsync(ct), timeout, cancellationToken)
+                    .ConfigureAwait(false);
+            }
+            catch (Exception exception)
+            {
+                if (IsWrappedException(exception))
+                {
+                    throw;
+                }
+
+                WrapException(exception);
+            }
+        }
+
+        public async Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        {
+            ThrowIfDisposed();
+
+            try
+            {
+                _logger.Verbose("Disconnecting [Timeout={0}]", timeout);
+
+                await Internal.TaskExtensions
+                    .TimeoutAfterAsync(ct => _channel.DisconnectAsync(), timeout, cancellationToken)
+                    .ConfigureAwait(false);
+            }
+            catch (Exception exception)
+            {
+                if (IsWrappedException(exception))
+                {
+                    throw;
+                }
+
+                WrapException(exception);
+            }
+        }
+
+        public async Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
+        {
+            await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+            try
+            {
+                _logger.Verbose("TX >>> {0}", packet);
 
                 var packetData = PacketSerializer.Serialize(packet);
 
-                return Internal.TaskExtensions.TimeoutAfter(ct => _channel.WriteAsync(
-                    packetData.Array,
-                    packetData.Offset,
-                    packetData.Count,
-                    ct), timeout, cancellationToken);
-            });
+                await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false);
+
+                PacketSerializer.FreeBuffer();
+            }
+            catch (Exception exception)
+            {
+                if (IsWrappedException(exception))
+                {
+                    throw;
+                }
+
+                WrapException(exception);
+            }
+            finally
+            {
+                _writerSemaphore.Release();
+            }
         }
 
         public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
 
-            MqttBasePacket packet = null;
-            await ExecuteAndWrapExceptionAsync(async () =>
+            try
             {
                 ReceivedMqttPacket receivedMqttPacket;
 
                 if (timeout > TimeSpan.Zero)
                 {
-                    receivedMqttPacket = await Internal.TaskExtensions.TimeoutAfter(ct => ReceiveAsync(_channel, ct), timeout, cancellationToken).ConfigureAwait(false);
+                    receivedMqttPacket = await Internal.TaskExtensions.TimeoutAfterAsync(ct => ReceiveAsync(_channel, ct), timeout, cancellationToken).ConfigureAwait(false);
                 }
                 else
                 {
@@ -94,28 +135,35 @@ namespace MQTTnet.Adapter
 
                 if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
                 {
-                    return;
+                    return null;
                 }
 
-                packet = PacketSerializer.Deserialize(receivedMqttPacket);
+                var packet = PacketSerializer.Deserialize(receivedMqttPacket);
                 if (packet == null)
                 {
                     throw new MqttProtocolViolationException("Received malformed packet.");
                 }
 
                 _logger.Verbose("RX <<< {0}", packet);
-            }).ConfigureAwait(false);
+                
+                return packet;
+            }
+            catch (Exception exception)
+            {
+                if (IsWrappedException(exception))
+                {
+                    throw;
+                }
 
-            return packet;
+                WrapException(exception);
+            }
+
+            return null;
         }
 
         private async Task<ReceivedMqttPacket> ReceiveAsync(IMqttChannel channel, CancellationToken cancellationToken)
         {
             var fixedHeader = await MqttPacketReader.ReadFixedHeaderAsync(channel, cancellationToken).ConfigureAwait(false);
-            if (fixedHeader == null)
-            {
-                return null;
-            }
 
             try
             {
@@ -138,7 +186,9 @@ namespace MQTTnet.Adapter
                         chunkSize = bytesLeft;
                     }
 
-                    var readBytes = await channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken) .ConfigureAwait(false);
+                    // async/await is not used to avoid the overhead of context switches. We assume that the reamining data
+                    // has been sent from the sender directly after the initial bytes.
+                    var readBytes = channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).GetAwaiter().GetResult();
                     if (readBytes <= 0)
                     {
                         ExceptionHelper.ThrowGracefulSocketClose();
@@ -147,7 +197,7 @@ namespace MQTTnet.Adapter
                     bodyOffset += readBytes;
                 } while (bodyOffset < body.Length);
 
-                return new ReceivedMqttPacket(fixedHeader.Flags, new MqttPacketBodyReader(body));
+                return new ReceivedMqttPacket(fixedHeader.Flags, new MqttPacketBodyReader(body, 0));
             }
             finally
             {
@@ -155,42 +205,6 @@ namespace MQTTnet.Adapter
             }
         }
 
-        private static async Task ExecuteAndWrapExceptionAsync(Func<Task> action)
-        {
-            try
-            {
-                await action().ConfigureAwait(false);
-            }
-            catch (Exception exception)
-            {
-                if (exception is TaskCanceledException ||
-                    exception is OperationCanceledException ||
-                    exception is MqttCommunicationTimedOutException ||
-                    exception is MqttCommunicationException)
-                {
-                    throw;
-                }
-
-                if (exception is IOException && exception.InnerException is SocketException socketException)
-                {
-                    if (socketException.SocketErrorCode == SocketError.ConnectionAborted)
-                    {
-                        throw new OperationCanceledException();
-                    }
-                }
-
-                if (exception is COMException comException)
-                {
-                    if ((uint)comException.HResult == ErrorOperationAborted)
-                    {
-                        throw new OperationCanceledException();
-                    }
-                }
-
-                throw new MqttCommunicationException(exception);
-            }
-        }
-
         public void Dispose()
         {
             _isDisposed = true;
@@ -205,5 +219,34 @@ namespace MQTTnet.Adapter
                 throw new ObjectDisposedException(nameof(MqttChannelAdapter));
             }
         }
+
+        private static bool IsWrappedException(Exception exception)
+        {
+            return exception is TaskCanceledException ||
+                   exception is OperationCanceledException ||
+                   exception is MqttCommunicationTimedOutException ||
+                   exception is MqttCommunicationException;
+        }
+
+        private static void WrapException(Exception exception)
+        {
+            if (exception is IOException && exception.InnerException is SocketException socketException)
+            {
+                if (socketException.SocketErrorCode == SocketError.ConnectionAborted)
+                {
+                    throw new OperationCanceledException();
+                }
+            }
+
+            if (exception is COMException comException)
+            {
+                if ((uint)comException.HResult == ErrorOperationAborted)
+                {
+                    throw new OperationCanceledException();
+                }
+            }
+
+            throw new MqttCommunicationException(exception);
+        }
     }
 }

Source/MQTTnet/Client/MqttClient.cs

@@ -17,7 +17,7 @@ namespace MQTTnet.Client
     {
         private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
         private readonly Stopwatch _sendTracker = new Stopwatch();
-        private readonly SemaphoreSlim _disconnectLock = new SemaphoreSlim(1, 1);
+        private readonly object _disconnectLock = new object();
         private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
 
         private readonly IMqttClientAdapterFactory _adapterFactory;
@@ -215,7 +215,7 @@ namespace MQTTnet.Client
 
         private async Task DisconnectInternalAsync(Task sender, Exception exception)
         {
-            await InitiateDisconnectAsync().ConfigureAwait(false);
+            InitiateDisconnect();
 
             var clientWasConnected = IsConnected;
             IsConnected = false;
@@ -249,45 +249,38 @@ namespace MQTTnet.Client
             }
         }
 
-        private async Task InitiateDisconnectAsync()
+        private void InitiateDisconnect()
         {
-            await _disconnectLock.WaitAsync().ConfigureAwait(false);
-            try
+            lock (_disconnectLock)
             {
-                if (_cancellationTokenSource == null || _cancellationTokenSource.IsCancellationRequested)
+                try
                 {
-                    return;
-                }
+                    if (_cancellationTokenSource == null || _cancellationTokenSource.IsCancellationRequested)
+                    {
+                        return;
+                    }
 
-                _cancellationTokenSource.Cancel(false);
-            }
-            catch (Exception adapterException)
-            {
-                _logger.Warning(adapterException, "Error while initiating disconnect.");
-            }
-            finally
-            {
-                _disconnectLock.Release();
+                    _cancellationTokenSource.Cancel(false);
+                }
+                catch (Exception adapterException)
+                {
+                    _logger.Warning(adapterException, "Error while initiating disconnect.");
+                }
             }
         }
 
         private Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken)
         {
-            if (cancellationToken.IsCancellationRequested)
-            {
-                throw new TaskCanceledException();
-            }
+            cancellationToken.ThrowIfCancellationRequested();
 
             _sendTracker.Restart();
-            return _adapter.SendPacketAsync(_options.CommunicationTimeout, packet, cancellationToken);
+
+            return _adapter.SendPacketAsync(packet, cancellationToken);
         }
 
         private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket
         {
-            if (cancellationToken.IsCancellationRequested)
-            {
-                throw new TaskCanceledException();
-            }
+            cancellationToken.ThrowIfCancellationRequested();
 
             _sendTracker.Restart();
 
@@ -300,8 +293,8 @@ namespace MQTTnet.Client
             var packetAwaiter = _packetDispatcher.AddPacketAwaiter<TResponsePacket>(identifier);
             try
             {
-                await _adapter.SendPacketAsync(_options.CommunicationTimeout, requestPacket, cancellationToken).ConfigureAwait(false);
-                var respone = await Internal.TaskExtensions.TimeoutAfter(ct => packetAwaiter.Task, _options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
+                await _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false);
+                var respone = await Internal.TaskExtensions.TimeoutAfterAsync(ct => packetAwaiter.Task, _options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
 
                 return (TResponsePacket)respone;
             }
@@ -526,7 +519,7 @@ namespace MQTTnet.Client
             {
                 await task.ConfigureAwait(false);
             }
-            catch (TaskCanceledException)
+            catch (OperationCanceledException)
             {
             }
         }

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/Diagnostics/TargetFrameworkInfoProvider.cs

@@ -10,6 +10,8 @@
                 return "net452";
 #elif NET461
                 return "net461";
+#elif NET472
+                return "net472";
 #elif NETSTANDARD1_3
                 return "netstandard1.3";
 #elif NETSTANDARD2_0

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/Implementations/MqttTcpChannel.cs

@@ -1,4 +1,4 @@
-#if NET452 || NET461 || NETSTANDARD1_3 || NETSTANDARD2_0
+#if !WINDOWS_UWP
 using System;
 using System.Net.Security;
 using System.Net.Sockets;

Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs

@@ -1,4 +1,4 @@
-#if NET452 || NET461 || NETSTANDARD1_3 || NETSTANDARD2_0
+#if !WINDOWS_UWP
 using System;
 using System.Collections.Generic;
 using System.Net.Sockets;

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -1,4 +1,4 @@
-#if NET452 || NET461 || NETSTANDARD1_3 || NETSTANDARD2_0
+#if !WINDOWS_UWP
 using System;
 using System.Net;
 using System.Net.Security;
@@ -78,7 +78,8 @@ namespace MQTTnet.Implementations
                         await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false);
                     }
 
-                    _logger.Verbose($"Client '{clientSocket.RemoteEndPoint}' accepted by TCP listener '{_socket.LocalEndPoint}, {_addressFamily}'.");
+                    var protocol = _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6";
+                    _logger.Verbose($"Client '{clientSocket.RemoteEndPoint}' accepted by TCP listener '{_socket.LocalEndPoint}, {protocol}'.");
 
                     var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _logger);
                     ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
@@ -104,7 +105,7 @@ namespace MQTTnet.Implementations
         {
             _socket?.Dispose();
 
-#if NETSTANDARD1_3 || NETSTANDARD2_0 || NET461
+#if NETSTANDARD1_3 || NETSTANDARD2_0 || NET461 || NET472
             _tlsCertificate?.Dispose();
 #endif
         }

Source/MQTTnet/Internal/AsyncAutoResetEvent.cs

@@ -11,8 +11,10 @@ namespace MQTTnet.Internal
         private readonly LinkedList<TaskCompletionSource<bool>> _waiters = new LinkedList<TaskCompletionSource<bool>>();
         private bool _isSignaled;
 
-        public AsyncAutoResetEvent() : this(false)
-        { }
+        public AsyncAutoResetEvent()
+            : this(false)
+        {
+        }
 
         public AsyncAutoResetEvent(bool signaled)
         {
@@ -58,27 +60,24 @@ namespace MQTTnet.Internal
             }
 
             var winner = await Task.WhenAny(tcs.Task, Task.Delay(timeout, cancellationToken)).ConfigureAwait(false);
-            if (winner == tcs.Task)
+            var taskWasSignaled = winner == tcs.Task;
+            if (taskWasSignaled)
             {
-                // The task was signaled.
                 return true;
             }
-            else
+
+            // We timed-out; remove our reference to the task.
+            // This is an O(n) operation since waiters is a LinkedList<T>.
+            lock (_waiters)
             {
-                // We timed-out; remove our reference to the task.
-                // This is an O(n) operation since waiters is a LinkedList<T>.
-                lock (_waiters)
+                _waiters.Remove(tcs);
+
+                if (winner.Status == TaskStatus.Canceled)
                 {
-                    _waiters.Remove(tcs);
-                    if (winner.Status == TaskStatus.Canceled)
-                    {
-                        throw new OperationCanceledException(cancellationToken);
-                    }
-                    else
-                    {
-                        throw new TimeoutException();
-                    }
+                    throw new OperationCanceledException(cancellationToken);
                 }
+
+                throw new TimeoutException();
             }
         }
 

Source/MQTTnet/Internal/AsyncLock.cs

@@ -17,7 +17,7 @@ namespace MQTTnet.Internal
 
         public Task<IDisposable> LockAsync(CancellationToken cancellationToken)
         {
-            Task wait = _semaphore.WaitAsync(cancellationToken);
+            var wait = _semaphore.WaitAsync(cancellationToken);
             return wait.IsCompleted ?
                         _releaser :
                         wait.ContinueWith((_, state) => (IDisposable)state,

Source/MQTTnet/Internal/TaskExtensions.cs

@@ -7,7 +7,7 @@ namespace MQTTnet.Internal
 {
     public static class TaskExtensions
     {
-        public static async Task TimeoutAfter(Func<CancellationToken, Task> action, TimeSpan timeout, CancellationToken cancellationToken)
+        public static async Task TimeoutAfterAsync(Func<CancellationToken, Task> action, TimeSpan timeout, CancellationToken cancellationToken)
         {
             if (action == null) throw new ArgumentNullException(nameof(action));
 
@@ -31,7 +31,7 @@ namespace MQTTnet.Internal
             }
         }
 
-        public static async Task<TResult> TimeoutAfter<TResult>(Func<CancellationToken, Task<TResult>> action, TimeSpan timeout, CancellationToken cancellationToken)
+        public static async Task<TResult> TimeoutAfterAsync<TResult>(Func<CancellationToken, Task<TResult>> action, TimeSpan timeout, CancellationToken cancellationToken)
         {
             if (action == null) throw new ArgumentNullException(nameof(action));
 

Source/MQTTnet/MqttFactory.cs

@@ -22,6 +22,14 @@ namespace MQTTnet
             return new MqttClient(new MqttClientAdapterFactory(), logger);
         }
 
+        public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory mqttClientAdapterFactory)
+        {
+            if (logger == null) throw new ArgumentNullException(nameof(logger));
+            if (mqttClientAdapterFactory == null) throw new ArgumentNullException(nameof(mqttClientAdapterFactory));
+
+            return new MqttClient(mqttClientAdapterFactory, logger);
+        }
+
         public IMqttServer CreateMqttServer()
         {
             var logger = new MqttNetLogger();

Source/MQTTnet/Serializer/ByteReader.cs

@@ -1,48 +0,0 @@
-using System;
-
-namespace MQTTnet.Serializer
-{
-    public class ByteReader
-    {
-        private readonly int _source;
-        private int _index;
-
-        public ByteReader(int source)
-        {
-            _source = source;
-        }
-
-        public bool Read()
-        {
-            if (_index >= 8)
-            {
-                throw new InvalidOperationException("End of byte reached.");
-            }
-
-            var result = ((1 << _index) & _source) > 0;
-            _index++;
-            return result;
-        }
-
-        public int Read(int count)
-        {
-            if (_index + count > 8)
-            {
-                throw new InvalidOperationException("End of byte will be reached.");
-            }
-
-            var result = 0;
-            for (var i = 0; i < count; i++)
-            {
-                if (((1 << _index) & _source) > 0)
-                {
-                    result |= 1 << i;
-                }
-
-                _index++;
-            }
-
-            return result;
-        }
-    }
-}

Source/MQTTnet/Serializer/ByteWriter.cs

@@ -1,36 +0,0 @@
-using System;
-
-namespace MQTTnet.Serializer
-{
-    public class ByteWriter
-    {
-        private int _index;
-        private int _byte;
-
-        public byte Value => (byte)_byte;
-
-        public void Write(int @byte, int count)
-        {
-            for (var i = 0; i < count; i++)
-            {
-                var value = ((1 << i) & @byte) > 0;
-                Write(value);
-            }
-        }
-
-        public void Write(bool bit)
-        {
-            if (_index >= 8)
-            {
-                throw new InvalidOperationException("End of the byte reached.");
-            }
-
-            if (bit)
-            {
-                _byte |= 1 << _index;
-            }
-
-            _index++;
-        }
-    }
-}

Source/MQTTnet/Serializer/Extensions.cs

@@ -12,7 +12,10 @@ namespace MQTTnet.Serializer
             }
 
             var buffer = new byte[source.Count];
-            Buffer.BlockCopy(source.Array, source.Offset, buffer, 0, buffer.Length);
+            if (buffer.Length > 0)
+            {
+                Array.Copy(source.Array, source.Offset, buffer, 0, buffer.Length);
+            }
 
             return buffer;
         }

Source/MQTTnet/Serializer/IMqttPacketSerializer.cs

@@ -11,5 +11,7 @@ namespace MQTTnet.Serializer
         ArraySegment<byte> Serialize(MqttBasePacket mqttPacket);
 
         MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket);
+
+        void FreeBuffer();
     }
 }

Source/MQTTnet/Serializer/MqttFixedHeader.cs

@@ -1,6 +1,6 @@
 namespace MQTTnet.Serializer
 {
-    public class MqttFixedHeader
+    public struct MqttFixedHeader
     {
         public MqttFixedHeader(byte flags, int remainingLength)
         {
@@ -10,6 +10,6 @@
 
         public byte Flags { get; }
 
-        public int RemainingLength { get; set; }
+        public int RemainingLength { get; }
     }
 }

Source/MQTTnet/Serializer/MqttPacketBodyReader.cs

@@ -8,9 +8,10 @@ namespace MQTTnet.Serializer
         private readonly byte[] _buffer;
         private int _offset;
 
-        public MqttPacketBodyReader(byte[] buffer)
+        public MqttPacketBodyReader(byte[] buffer, int offset)
         {
             _buffer = buffer;
+            _offset = offset;
         }
 
         public int Length => _buffer.Length - _offset;

Source/MQTTnet/Serializer/MqttPacketReader.cs

@@ -1,4 +1,5 @@
-using System.Threading;
+using System;
+using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Channel;
 using MQTTnet.Exceptions;
@@ -8,23 +9,27 @@ namespace MQTTnet.Serializer
 {
     public static class MqttPacketReader
     {
+        [ThreadStatic]
+        private static byte[] _fixedHeaderBuffer;
+
+        [ThreadStatic]
+        private static byte[] _singleByteBuffer;
+
         public static async Task<MqttFixedHeader> ReadFixedHeaderAsync(IMqttChannel channel, CancellationToken cancellationToken)
         {
             // The MQTT fixed header contains 1 byte of flags and at least 1 byte for the remaining data length.
             // So in all cases at least 2 bytes must be read for a complete MQTT packet.
-            var buffer = new byte[2];
+            // async/await is used here because the next packet is received in a couple of minutes so the performance
+            // impact is acceptable according to a useless waiting thread.
+            var buffer = InitializeFixedHeaderBuffer();
             var totalBytesRead = 0;
 
             while (totalBytesRead < buffer.Length)
             {
-                var bytesRead = await channel.ReadAsync(buffer, 0, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
+                var bytesRead = await channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
                 if (bytesRead <= 0)
                 {
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        return null;
-                    }
-
+                    cancellationToken.ThrowIfCancellationRequested();
                     ExceptionHelper.ThrowGracefulSocketClose();
                 }
 
@@ -37,11 +42,11 @@ namespace MQTTnet.Serializer
                 return new MqttFixedHeader(buffer[0], 0);
             }
 
-            var bodyLength = await ReadBodyLengthAsync(channel, buffer[1], cancellationToken).ConfigureAwait(false);
+            var bodyLength = ReadBodyLength(channel, buffer[1], cancellationToken);
             return new MqttFixedHeader(buffer[0], bodyLength);
         }
 
-        private static async Task<int> ReadBodyLengthAsync(IMqttChannel channel, byte initialEncodedByte, CancellationToken cancellationToken)
+        private static int ReadBodyLength(IMqttChannel channel, byte initialEncodedByte, CancellationToken cancellationToken)
         {
             // Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html.
             var multiplier = 128;
@@ -50,7 +55,13 @@ namespace MQTTnet.Serializer
 
             while ((encodedByte & 128) != 0)
             {
-                encodedByte = await ReadByteAsync(channel, cancellationToken).ConfigureAwait(false);
+                cancellationToken.ThrowIfCancellationRequested();
+
+                // Here the async/await pattern is not used becuase the overhead of context switches
+                // is too big for reading 1 byte in a row. We expect that the remaining data was sent
+                // directly after the initial bytes. If the client disconnects just in this moment we
+                // will get an exception anyway.
+                encodedByte = ReadByte(channel, cancellationToken);
 
                 value += (byte)(encodedByte & 127) * multiplier;
                 if (multiplier > 128 * 128 * 128)
@@ -64,16 +75,27 @@ namespace MQTTnet.Serializer
             return value;
         }
 
-        private static async Task<byte> ReadByteAsync(IMqttChannel channel, CancellationToken cancellationToken)
+        private static byte ReadByte(IMqttChannel channel, CancellationToken cancellationToken)
         {
-            var buffer = new byte[1];
-            var readCount = await channel.ReadAsync(buffer, 0, 1, cancellationToken).ConfigureAwait(false);
+            var buffer = InitializeSingleByteBuffer();
+            var readCount = channel.ReadAsync(buffer, 0, 1, cancellationToken).GetAwaiter().GetResult();
             if (readCount <= 0)
             {
+                cancellationToken.ThrowIfCancellationRequested();
                 ExceptionHelper.ThrowGracefulSocketClose();
             }
 
             return buffer[0];
         }
+
+        private static byte[] InitializeFixedHeaderBuffer()
+        {
+            return _fixedHeaderBuffer ?? (_fixedHeaderBuffer = new byte[2]);
+        }
+
+        private static byte[] InitializeSingleByteBuffer()
+        {
+            return _singleByteBuffer ?? (_singleByteBuffer = new byte[1]);
+        }
     }
 }

Source/MQTTnet/Serializer/MqttPacketSerializer.cs

@@ -2,7 +2,6 @@
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using System;
-using System.IO;
 using System.Linq;
 using MQTTnet.Adapter;
 
@@ -12,65 +11,33 @@ namespace MQTTnet.Serializer
     {
         private const int FixedHeaderSize = 1;
 
+        private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter();
+
         public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
 
         public ArraySegment<byte> Serialize(MqttBasePacket packet)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
-            using (var stream = new MemoryStream(128))
-            {
-                // Leave enough head space for max header size (fixed + 4 variable remaining length = 5 bytes)
-                stream.Seek(5, SeekOrigin.Begin);
+            // Leave enough head space for max header size (fixed + 4 variable remaining length = 5 bytes)
+            _packetWriter.Reset();
+            _packetWriter.Seek(5);
 
-                var fixedHeader = SerializePacket(packet, stream);
-                var remainingLength = (int)stream.Length - 5;
+            var fixedHeader = SerializePacket(packet, _packetWriter);
+            var remainingLength = _packetWriter.Length - 5;
 
-                var remainingLengthBuffer = MqttPacketWriter.EncodeRemainingLength(remainingLength);
+            var remainingLengthBuffer = MqttPacketWriter.EncodeRemainingLength(remainingLength);
 
-                var headerSize = FixedHeaderSize + remainingLengthBuffer.Count;
-                var headerOffset = 5 - headerSize;
+            var headerSize = FixedHeaderSize + remainingLengthBuffer.Count;
+            var headerOffset = 5 - headerSize;
 
-                // Position cursor on correct offset on beginining of array (has leading 0x0)
-                stream.Seek(headerOffset, SeekOrigin.Begin);
-                stream.WriteByte(fixedHeader);
-                stream.Write(remainingLengthBuffer.Array, remainingLengthBuffer.Offset, remainingLengthBuffer.Count);
+            // 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);
 
-#if NET461 || NET452 || NETSTANDARD2_0
-                var buffer = stream.GetBuffer();
-                return new ArraySegment<byte>(buffer, headerOffset, (int)stream.Length - headerOffset);
-#else
-                if (stream.TryGetBuffer(out var segment))
-                {
-                    return new ArraySegment<byte>(segment.Array, headerOffset, segment.Count - headerOffset);
-                }
-
-                var buffer = stream.ToArray();
-                return new ArraySegment<byte>(buffer, headerOffset, buffer.Length - headerOffset);
-#endif          
-            }
-        }
-
-        private byte SerializePacket(MqttBasePacket packet, Stream stream)
-        {
-            switch (packet)
-            {
-                case MqttConnectPacket connectPacket: return Serialize(connectPacket, stream);
-                case MqttConnAckPacket connAckPacket: return Serialize(connAckPacket, stream);
-                case MqttDisconnectPacket _: return SerializeEmptyPacket(MqttControlPacketType.Disconnect);
-                case MqttPingReqPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingReq);
-                case MqttPingRespPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingResp);
-                case MqttPublishPacket publishPacket: return Serialize(publishPacket, stream);
-                case MqttPubAckPacket pubAckPacket: return Serialize(pubAckPacket, stream);
-                case MqttPubRecPacket pubRecPacket: return Serialize(pubRecPacket, stream);
-                case MqttPubRelPacket pubRelPacket: return Serialize(pubRelPacket, stream);
-                case MqttPubCompPacket pubCompPacket: return Serialize(pubCompPacket, stream);
-                case MqttSubscribePacket subscribePacket: return Serialize(subscribePacket, stream);
-                case MqttSubAckPacket subAckPacket: return Serialize(subAckPacket, stream);
-                case MqttUnsubscribePacket unsubscribePacket: return Serialize(unsubscribePacket, stream);
-                case MqttUnsubAckPacket unsubAckPacket: return Serialize(unsubAckPacket, stream);
-                default: throw new MqttProtocolViolationException("Packet type invalid.");
-            }
+            var buffer = _packetWriter.GetBuffer();
+            return new ArraySegment<byte>(buffer, headerOffset, _packetWriter.Length - headerOffset);
         }
 
         public MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket)
@@ -104,6 +71,33 @@ namespace MQTTnet.Serializer
             }
         }
 
+        public void FreeBuffer()
+        {
+            _packetWriter.FreeBuffer();
+        }
+
+        private byte SerializePacket(MqttBasePacket packet, MqttPacketWriter packetWriter)
+        {
+            switch (packet)
+            {
+                case MqttConnectPacket connectPacket: return Serialize(connectPacket, packetWriter);
+                case MqttConnAckPacket connAckPacket: return Serialize(connAckPacket, packetWriter);
+                case MqttDisconnectPacket _: return SerializeEmptyPacket(MqttControlPacketType.Disconnect);
+                case MqttPingReqPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingReq);
+                case MqttPingRespPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingResp);
+                case MqttPublishPacket publishPacket: return Serialize(publishPacket, packetWriter);
+                case MqttPubAckPacket pubAckPacket: return Serialize(pubAckPacket, packetWriter);
+                case MqttPubRecPacket pubRecPacket: return Serialize(pubRecPacket, packetWriter);
+                case MqttPubRelPacket pubRelPacket: return Serialize(pubRelPacket, packetWriter);
+                case MqttPubCompPacket pubCompPacket: return Serialize(pubCompPacket, packetWriter);
+                case MqttSubscribePacket subscribePacket: return Serialize(subscribePacket, packetWriter);
+                case MqttSubAckPacket subAckPacket: return Serialize(subAckPacket, packetWriter);
+                case MqttUnsubscribePacket unsubscribePacket: return Serialize(unsubscribePacket, packetWriter);
+                case MqttUnsubAckPacket unsubAckPacket: return Serialize(unsubAckPacket, packetWriter);
+                default: throw new MqttProtocolViolationException("Packet type invalid.");
+            }
+        }
+
         private static MqttBasePacket DeserializeUnsubAck(MqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
@@ -192,20 +186,18 @@ namespace MQTTnet.Serializer
 
         private static MqttBasePacket DeserializePublish(ReceivedMqttPacket receivedMqttPacket)
         {
-            var body = receivedMqttPacket.Body;
-            ThrowIfBodyIsEmpty(body);
+            ThrowIfBodyIsEmpty(receivedMqttPacket.Body);
 
-            var fixedHeader = new ByteReader(receivedMqttPacket.FixedHeader);
-            var retain = fixedHeader.Read();
-            var qualityOfServiceLevel = (MqttQualityOfServiceLevel)fixedHeader.Read(2);
-            var dup = fixedHeader.Read();
+            var retain = (receivedMqttPacket.FixedHeader & 0x1) > 0;
+            var qualityOfServiceLevel = (MqttQualityOfServiceLevel)(receivedMqttPacket.FixedHeader >> 1 & 0x3);
+            var dup = (receivedMqttPacket.FixedHeader & 0x3) > 0;
 
-            var topic = body.ReadStringWithLengthPrefix();
+            var topic = receivedMqttPacket.Body.ReadStringWithLengthPrefix();
 
             ushort? packetIdentifier = null;
             if (qualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
             {
-                packetIdentifier = body.ReadUInt16();
+                packetIdentifier = receivedMqttPacket.Body.ReadUInt16();
             }
 
             var packet = new MqttPublishPacket
@@ -213,7 +205,7 @@ namespace MQTTnet.Serializer
                 PacketIdentifier = packetIdentifier,
                 Retain = retain,
                 Topic = topic,
-                Payload = body.ReadRemainingData().ToArray(),
+                Payload = receivedMqttPacket.Body.ReadRemainingData().ToArray(),
                 QualityOfServiceLevel = qualityOfServiceLevel,
                 Dup = dup
             };
@@ -253,8 +245,8 @@ namespace MQTTnet.Serializer
                 throw new MqttProtocolViolationException($"Protocol name ({protocolName}) is not supported.");
             }
 
-            var connectFlags = new ByteReader(body.ReadByte());
-            if (connectFlags.Read())
+            var connectFlags = body.ReadByte();
+            if ((connectFlags & 0x1) > 0)
             {
                 throw new MqttProtocolViolationException("The first bit of the Connect Flags must be set to 0.");
             }
@@ -262,14 +254,14 @@ namespace MQTTnet.Serializer
             var packet = new MqttConnectPacket
             {
                 ProtocolVersion = protocolVersion,
-                CleanSession = connectFlags.Read()
+                CleanSession = (connectFlags & 0x2) > 0
             };
 
-            var willFlag = connectFlags.Read();
-            var willQoS = connectFlags.Read(2);
-            var willRetain = connectFlags.Read();
-            var passwordFlag = connectFlags.Read();
-            var usernameFlag = connectFlags.Read();
+            var willFlag = (connectFlags & 0x4) > 0;
+            var willQoS = (connectFlags & 0x18) >> 3;
+            var willRetain = (connectFlags & 0x20) > 0;
+            var passwordFlag = (connectFlags & 0x40) > 0;
+            var usernameFlag = (connectFlags & 0x80) > 0;
 
             packet.KeepAlivePeriod = body.ReadUInt16();
             packet.ClientId = body.ReadStringWithLengthPrefix();
@@ -322,11 +314,11 @@ namespace MQTTnet.Serializer
 
             var packet = new MqttConnAckPacket();
 
-            var firstByteReader = new ByteReader(body.ReadByte());
+            var acknowledgeFlags = body.ReadByte();
 
             if (ProtocolVersion == MqttProtocolVersion.V311)
             {
-                packet.IsSessionPresent = firstByteReader.Read();
+                packet.IsSessionPresent = (acknowledgeFlags & 0x1) > 0;
             }
 
             packet.ConnectReturnCode = (MqttConnectReturnCode)body.ReadByte();
@@ -344,46 +336,46 @@ namespace MQTTnet.Serializer
             }
         }
 
+        // ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local
         private static void ValidatePublishPacket(MqttPublishPacket packet)
         {
-            if (packet == null) throw new ArgumentNullException(nameof(packet));
-
             if (packet.QualityOfServiceLevel == 0 && packet.Dup)
             {
                 throw new MqttProtocolViolationException("Dup flag must be false for QoS 0 packets [MQTT-3.3.1-2].");
             }
         }
 
-        private byte Serialize(MqttConnectPacket packet, Stream stream)
+        private byte Serialize(MqttConnectPacket packet, MqttPacketWriter packetWriter)
         {
             ValidateConnectPacket(packet);
 
             // Write variable header
             if (ProtocolVersion == MqttProtocolVersion.V311)
             {
-                stream.WriteWithLengthPrefix("MQTT");
-                stream.WriteByte(4); // 3.1.2.2 Protocol Level 4
+                packetWriter.WriteWithLengthPrefix("MQTT");
+                packetWriter.Write(4); // 3.1.2.2 Protocol Level 4
             }
             else
             {
-                stream.WriteWithLengthPrefix("MQIsdp");
-                stream.WriteByte(3); // Protocol Level 3
+                packetWriter.WriteWithLengthPrefix("MQIsdp");
+                packetWriter.Write(3); // Protocol Level 3
             }
 
-            var connectFlags = new ByteWriter(); // 3.1.2.3 Connect Flags
-            connectFlags.Write(false); // Reserved
-            connectFlags.Write(packet.CleanSession);
-            connectFlags.Write(packet.WillMessage != null);
+            byte connectFlags = 0x0;
+            if (packet.CleanSession)
+            {
+                connectFlags |= 0x2;
+            }
 
             if (packet.WillMessage != null)
             {
-                connectFlags.Write((int)packet.WillMessage.QualityOfServiceLevel, 2);
-                connectFlags.Write(packet.WillMessage.Retain);
-            }
-            else
-            {
-                connectFlags.Write(0, 2);
-                connectFlags.Write(false);
+                connectFlags |= 0x4;
+                connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3);
+
+                if (packet.WillMessage.Retain)
+                {
+                    connectFlags |= 0x20;
+                }
             }
 
             if (packet.Password != null && packet.Username == null)
@@ -391,72 +383,82 @@ namespace MQTTnet.Serializer
                 throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22].");
             }
 
-            connectFlags.Write(packet.Password != null);
-            connectFlags.Write(packet.Username != null);
-
-            stream.Write(connectFlags);
-            stream.Write(packet.KeepAlivePeriod);
-            stream.WriteWithLengthPrefix(packet.ClientId);
-
-            if (packet.WillMessage != null)
+            if (packet.Password != null)
             {
-                stream.WriteWithLengthPrefix(packet.WillMessage.Topic);
-                stream.WriteWithLengthPrefix(packet.WillMessage.Payload);
+                connectFlags |= 0x40;
             }
 
             if (packet.Username != null)
             {
-                stream.WriteWithLengthPrefix(packet.Username);
+                connectFlags |= 0x80;
+            }
+
+            packetWriter.Write(connectFlags);
+            packetWriter.Write(packet.KeepAlivePeriod);
+            packetWriter.WriteWithLengthPrefix(packet.ClientId);
+
+            if (packet.WillMessage != null)
+            {
+                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic);
+                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload);
+            }
+
+            if (packet.Username != null)
+            {
+                packetWriter.WriteWithLengthPrefix(packet.Username);
             }
 
             if (packet.Password != null)
             {
-                stream.WriteWithLengthPrefix(packet.Password);
+                packetWriter.WriteWithLengthPrefix(packet.Password);
             }
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect);
         }
 
-        private byte Serialize(MqttConnAckPacket packet, Stream stream)
+        private byte Serialize(MqttConnAckPacket packet, MqttPacketWriter packetWriter)
         {
             if (ProtocolVersion == MqttProtocolVersion.V310)
             {
-                stream.WriteByte(0);
+                packetWriter.Write(0);
             }
             else if (ProtocolVersion == MqttProtocolVersion.V311)
             {
-                var connectAcknowledgeFlags = new ByteWriter();
-                connectAcknowledgeFlags.Write(packet.IsSessionPresent);
+                byte connectAcknowledgeFlags = 0x0;
+                if (packet.IsSessionPresent)
+                {
+                    connectAcknowledgeFlags |= 0x1;
+                }
 
-                stream.Write(connectAcknowledgeFlags);
+                packetWriter.Write(connectAcknowledgeFlags);
             }
             else
             {
                 throw new MqttProtocolViolationException("Protocol version not supported.");
             }
 
-            stream.WriteByte((byte)packet.ConnectReturnCode);
+            packetWriter.Write((byte)packet.ConnectReturnCode);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
         }
 
-        private static byte Serialize(MqttPubRelPacket packet, Stream stream)
+        private static byte Serialize(MqttPubRelPacket packet, MqttPacketWriter packetWriter)
         {
             if (!packet.PacketIdentifier.HasValue)
             {
                 throw new MqttProtocolViolationException("PubRel packet has no packet identifier.");
             }
 
-            stream.Write(packet.PacketIdentifier.Value);
+            packetWriter.Write(packet.PacketIdentifier.Value);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
         }
 
-        private static byte Serialize(MqttPublishPacket packet, Stream stream)
+        private static byte Serialize(MqttPublishPacket packet, MqttPacketWriter packetWriter)
         {
             ValidatePublishPacket(packet);
 
-            stream.WriteWithLengthPrefix(packet.Topic);
+            packetWriter.WriteWithLengthPrefix(packet.Topic);
 
             if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
             {
@@ -465,7 +467,7 @@ namespace MQTTnet.Serializer
                     throw new MqttProtocolViolationException("Publish packet has no packet identifier.");
                 }
 
-                stream.Write(packet.PacketIdentifier.Value);
+                packetWriter.Write(packet.PacketIdentifier.Value);
             }
             else
             {
@@ -477,7 +479,7 @@ namespace MQTTnet.Serializer
 
             if (packet.Payload?.Length > 0)
             {
-                stream.Write(packet.Payload, 0, packet.Payload.Length);
+                packetWriter.Write(packet.Payload, 0, packet.Payload.Length);
             }
 
             byte fixedHeader = 0;
@@ -497,43 +499,43 @@ namespace MQTTnet.Serializer
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader);
         }
 
-        private static byte Serialize(MqttPubAckPacket packet, Stream stream)
+        private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter packetWriter)
         {
             if (!packet.PacketIdentifier.HasValue)
             {
                 throw new MqttProtocolViolationException("PubAck packet has no packet identifier.");
             }
 
-            stream.Write(packet.PacketIdentifier.Value);
+            packetWriter.Write(packet.PacketIdentifier.Value);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
         }
 
-        private static byte Serialize(MqttPubRecPacket packet, Stream stream)
+        private static byte Serialize(MqttPubRecPacket packet, MqttPacketWriter packetWriter)
         {
             if (!packet.PacketIdentifier.HasValue)
             {
                 throw new MqttProtocolViolationException("PubRec packet has no packet identifier.");
             }
 
-            stream.Write(packet.PacketIdentifier.Value);
+            packetWriter.Write(packet.PacketIdentifier.Value);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
         }
 
-        private static byte Serialize(MqttPubCompPacket packet, Stream stream)
+        private static byte Serialize(MqttPubCompPacket packet, MqttPacketWriter packetWriter)
         {
             if (!packet.PacketIdentifier.HasValue)
             {
                 throw new MqttProtocolViolationException("PubComp packet has no packet identifier.");
             }
 
-            stream.Write(packet.PacketIdentifier.Value);
+            packetWriter.Write(packet.PacketIdentifier.Value);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
         }
 
-        private static byte Serialize(MqttSubscribePacket packet, Stream stream)
+        private static byte Serialize(MqttSubscribePacket packet, MqttPacketWriter packetWriter)
         {
             if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
 
@@ -542,41 +544,41 @@ namespace MQTTnet.Serializer
                 throw new MqttProtocolViolationException("Subscribe packet has no packet identifier.");
             }
 
-            stream.Write(packet.PacketIdentifier.Value);
+            packetWriter.Write(packet.PacketIdentifier.Value);
 
             if (packet.TopicFilters?.Count > 0)
             {
                 foreach (var topicFilter in packet.TopicFilters)
                 {
-                    stream.WriteWithLengthPrefix(topicFilter.Topic);
-                    stream.WriteByte((byte)topicFilter.QualityOfServiceLevel);
+                    packetWriter.WriteWithLengthPrefix(topicFilter.Topic);
+                    packetWriter.Write((byte)topicFilter.QualityOfServiceLevel);
                 }
             }
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02);
         }
 
-        private static byte Serialize(MqttSubAckPacket packet, Stream stream)
+        private static byte Serialize(MqttSubAckPacket packet, MqttPacketWriter packetWriter)
         {
             if (!packet.PacketIdentifier.HasValue)
             {
                 throw new MqttProtocolViolationException("SubAck packet has no packet identifier.");
             }
 
-            stream.Write(packet.PacketIdentifier.Value);
+            packetWriter.Write(packet.PacketIdentifier.Value);
 
             if (packet.SubscribeReturnCodes?.Any() == true)
             {
                 foreach (var packetSubscribeReturnCode in packet.SubscribeReturnCodes)
                 {
-                    stream.WriteByte((byte)packetSubscribeReturnCode);
+                    packetWriter.Write((byte)packetSubscribeReturnCode);
                 }
             }
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck);
         }
 
-        private static byte Serialize(MqttUnsubscribePacket packet, Stream stream)
+        private static byte Serialize(MqttUnsubscribePacket packet, MqttPacketWriter packetWriter)
         {
             if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
 
@@ -585,27 +587,27 @@ namespace MQTTnet.Serializer
                 throw new MqttProtocolViolationException("Unsubscribe packet has no packet identifier.");
             }
 
-            stream.Write(packet.PacketIdentifier.Value);
+            packetWriter.Write(packet.PacketIdentifier.Value);
 
             if (packet.TopicFilters?.Any() == true)
             {
                 foreach (var topicFilter in packet.TopicFilters)
                 {
-                    stream.WriteWithLengthPrefix(topicFilter);
+                    packetWriter.WriteWithLengthPrefix(topicFilter);
                 }
             }
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
         }
 
-        private static byte Serialize(MqttUnsubAckPacket packet, Stream stream)
+        private static byte Serialize(MqttUnsubAckPacket packet, MqttPacketWriter packetWriter)
         {
             if (!packet.PacketIdentifier.HasValue)
             {
                 throw new MqttProtocolViolationException("UnsubAck packet has no packet identifier.");
             }
 
-            stream.Write(packet.PacketIdentifier.Value);
+            packetWriter.Write(packet.PacketIdentifier.Value);
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck);
         }
 
@@ -614,6 +616,7 @@ namespace MQTTnet.Serializer
             return MqttPacketWriter.BuildFixedHeader(type);
         }
 
+        // ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local
         private static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body)
         {
             if (body == null || body.Length == 0)

Source/MQTTnet/Serializer/MqttPacketWriter.cs

@@ -1,12 +1,25 @@
 using System;
-using System.IO;
 using System.Text;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Serializer
 {
-    public static class MqttPacketWriter
+    /// <summary>
+    /// This is a custom implementation of a memory stream which provides only MQTTnet relevant features.
+    /// The goal is to avoid lots of argument checks like in the original stream. The growth rule is the
+    /// same as for the original MemoryStream in .net. Also this implementation allows accessing the internal
+    /// buffer for all platforms and .net framework versions (which is not available at the regular MemoryStream).
+    /// </summary>
+    public class MqttPacketWriter
     {
+        public static int MaxBufferSize = 4096;
+
+        private byte[] _buffer = new byte[128];
+
+        private int _position;
+
+        public int Length { get; private set; }
+
         public static byte BuildFixedHeader(MqttControlPacketType packetType, byte flags = 0)
         {
             var fixedHeader = (int)packetType << 4;
@@ -14,33 +27,6 @@ namespace MQTTnet.Serializer
             return (byte)fixedHeader;
         }
 
-        public static void Write(this Stream stream, ushort value)
-        {
-            var buffer = BitConverter.GetBytes(value);
-            stream.WriteByte(buffer[1]);
-            stream.WriteByte(buffer[0]);
-        }
-
-        public static void Write(this Stream stream, ByteWriter value)
-        {
-            if (value == null) throw new ArgumentNullException(nameof(value));
-
-            stream.WriteByte(value.Value);
-        }
-
-        public static void WriteWithLengthPrefix(this Stream stream, string value)
-        {
-            stream.WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value ?? string.Empty));
-        }
-
-        public static void WriteWithLengthPrefix(this Stream stream, byte[] value)
-        {
-            var length = (ushort)value.Length;
-
-            stream.Write(length);
-            stream.Write(value, 0, length);
-        }
-
         public static ArraySegment<byte> EncodeRemainingLength(int length)
         {
             // write the encoded remaining length right aligned on the 4 byte buffer
@@ -69,5 +55,112 @@ namespace MQTTnet.Serializer
 
             return new ArraySegment<byte>(buffer, 0, bufferOffset);
         }
+
+        public void WriteWithLengthPrefix(string value)
+        {
+            WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value ?? string.Empty));
+        }
+
+        public void WriteWithLengthPrefix(byte[] value)
+        {
+            EnsureAdditionalCapacity(value.Length + 2);
+
+            Write((ushort)value.Length);
+            Write(value, 0, value.Length);
+        }
+        
+        public void Write(byte @byte)
+        {
+            EnsureAdditionalCapacity(1);
+
+            _buffer[_position] = @byte;
+            IncreasePostition(1);
+        }
+
+        public void Write(ushort value)
+        {
+            EnsureAdditionalCapacity(2);
+
+            _buffer[_position] = (byte)(value >> 8);
+            IncreasePostition(1);
+            _buffer[_position] = (byte)value;
+            IncreasePostition(1);
+        }
+
+        public void Write(byte[] array, int offset, int count)
+        {
+            EnsureAdditionalCapacity(count);
+
+            Array.Copy(array, offset, _buffer, _position, count);
+            IncreasePostition(count);
+        }
+
+        public void Reset()
+        {
+            Length = 0;
+        }
+
+        public void Seek(int offset)
+        {
+            EnsureCapacity(offset);
+            _position = offset;
+        }
+
+        public byte[] GetBuffer()
+        {
+            return _buffer;
+        }
+
+        public void FreeBuffer()
+        {
+            // This method frees the used memory by shrinking the buffer. This is required because the buffer
+            // is used across several messages. In general this is not a big issue because subsequent Ping packages
+            // have the same size but a very big publish package with 100 MB of payload will increase the buffer 
+            // 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 < MaxBufferSize)
+            {
+                return;
+            }
+
+            Array.Resize(ref _buffer, MaxBufferSize);
+        }
+
+        private void EnsureAdditionalCapacity(int additionalCapacity)
+        {
+            var freeSpace = _buffer.Length - _position;
+            if (freeSpace >= additionalCapacity)
+            {
+                return;
+            }
+
+            EnsureCapacity(additionalCapacity - freeSpace);
+        }
+
+        private void EnsureCapacity(int capacity)
+        {
+            if (_buffer.Length >= capacity)
+            {
+                return;
+            }
+
+            var newBufferLength = _buffer.Length;
+            while (newBufferLength < capacity)
+            {
+                newBufferLength *= 2;
+            }
+
+            Array.Resize(ref _buffer, newBufferLength);
+        }
+
+        private void IncreasePostition(int length)
+        {
+            _position += length;
+            if (_position > Length)
+            {
+                Length = _position;
+            }
+        }
     }
 }

Source/MQTTnet/Server/IMqttClientSession.cs

@@ -0,0 +1,23 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using MQTTnet.Adapter;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Server
+{
+    public interface IMqttClientSession : IDisposable
+    {
+        string ClientId { get; }
+        void FillStatus(MqttClientSessionStatus status);
+
+        void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket);
+        void ClearPendingApplicationMessages();
+        
+        Task<bool> RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter);
+        void Stop(MqttClientDisconnectType disconnectType);
+
+        Task SubscribeAsync(IList<TopicFilter> topicFilters);
+        Task UnsubscribeAsync(IList<string> topicFilters);
+    }
+}

Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs

@@ -12,19 +12,17 @@ namespace MQTTnet.Server
         private readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch();
         private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = new Stopwatch();
 
+        private readonly IMqttClientSession _clientSession;
         private readonly IMqttNetChildLogger _logger;
-        private readonly string _clientId;
-        private readonly Action _callback;
-
+        
         private bool _isPaused;
-        private Task _workerTask;
-
-        public MqttClientKeepAliveMonitor(string clientId, Action callback, IMqttNetChildLogger logger)
+        
+        public MqttClientKeepAliveMonitor(IMqttClientSession clientSession, IMqttNetChildLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
-            _clientId = clientId;
-            _callback = callback;
+            _clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
+
             _logger = logger.CreateChildLogger(nameof(MqttClientKeepAliveMonitor));
         }
 
@@ -39,7 +37,7 @@ namespace MQTTnet.Server
                 return;
             }
 
-            _workerTask = Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken);
+            Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken);
         }
 
         public void Pause()
@@ -74,9 +72,9 @@ namespace MQTTnet.Server
                     // Values described here: [MQTT-3.1.2-24].
                     if (!_isPaused && _lastPacketReceivedTracker.Elapsed.TotalSeconds > keepAlivePeriod * 1.5D)
                     {
-                        _logger.Warning(null, "Client '{0}': Did not receive any packet or keep alive signal.", _clientId);
-                        _callback();
-    
+                        _logger.Warning(null, "Client '{0}': Did not receive any packet or keep alive signal.", _clientSession.ClientId);
+                        _clientSession.Stop(MqttClientDisconnectType.NotClean);
+                        
                         return;
                     }
 
@@ -88,11 +86,11 @@ namespace MQTTnet.Server
             }
             catch (Exception exception)
             {
-                _logger.Error(exception, "Client '{0}': Unhandled exception while checking keep alive timeouts.", _clientId);
+                _logger.Error(exception, "Client '{0}': Unhandled exception while checking keep alive timeouts.", _clientSession.ClientId);
             }
             finally
             {
-                _logger.Verbose("Client {0}: Stopped checking keep alive timeout.", _clientId);
+                _logger.Verbose("Client {0}: Stopped checking keep alive timeout.", _clientSession.ClientId);
             }
         }
     }

Source/MQTTnet/Server/MqttClientPendingPacketsQueue.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
@@ -11,25 +11,34 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
-    public class MqttClientPendingMessagesQueue : IDisposable
+    public class MqttClientPendingPacketsQueue : IDisposable
     {
+        private readonly Queue<MqttBasePacket> _queue = new Queue<MqttBasePacket>();
         private readonly AsyncAutoResetEvent _queueAutoResetEvent = new AsyncAutoResetEvent();
+
         private readonly IMqttServerOptions _options;
         private readonly MqttClientSession _clientSession;
         private readonly IMqttNetChildLogger _logger;
 
-        private ConcurrentQueue<MqttBasePacket> _queue = new ConcurrentQueue<MqttBasePacket>();
-
-        public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetChildLogger logger)
+        public MqttClientPendingPacketsQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetChildLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
             _options = options ?? throw new ArgumentNullException(nameof(options));
             _clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
 
-            _logger = logger.CreateChildLogger(nameof(MqttClientPendingMessagesQueue));
+            _logger = logger.CreateChildLogger(nameof(MqttClientPendingPacketsQueue));
         }
 
-        public int Count => _queue.Count;
+        public int Count
+        {
+            get
+            {
+                lock (_queue)
+                {
+                    return _queue.Count;
+                }
+            }
+        }
 
         public void Start(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
         {
@@ -42,25 +51,29 @@ namespace MQTTnet.Server
 
             Task.Run(() => SendQueuedPacketsAsync(adapter, cancellationToken), cancellationToken);
         }
-        
+
         public void Enqueue(MqttBasePacket packet)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
-            if (_queue.Count >= _options.MaxPendingMessagesPerClient)
+            lock (_queue)
             {
-                if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
+                if (_queue.Count >= _options.MaxPendingMessagesPerClient)
                 {
-                    return;
-                }
+                    if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
+                    {
+                        return;
+                    }
 
-                if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
-                {
-                    _queue.TryDequeue(out _);
+                    if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
+                    {
+                        _queue.Dequeue();
+                    }
                 }
+                
+                _queue.Enqueue(packet);
             }
 
-            _queue.Enqueue(packet);
             _queueAutoResetEvent.Set();
 
             _logger.Verbose("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
@@ -68,13 +81,14 @@ namespace MQTTnet.Server
 
         public void Clear()
         {
-            var newQueue = new ConcurrentQueue<MqttBasePacket>();
-            Interlocked.Exchange(ref _queue, newQueue);
+            lock (_queue)
+            {
+                _queue.Clear();
+            }
         }
 
         public void Dispose()
         {
-            
         }
 
         private async Task SendQueuedPacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
@@ -100,13 +114,17 @@ namespace MQTTnet.Server
             MqttBasePacket packet = null;
             try
             {
-                if (_queue.IsEmpty)
+                lock (_queue)
                 {
-                    await _queueAutoResetEvent.WaitOneAsync(cancellationToken).ConfigureAwait(false);
+                    if (_queue.Count > 0)
+                    {
+                        packet = _queue.Dequeue();
+                    }
                 }
 
-                if (!_queue.TryDequeue(out packet))
+                if (packet == null)
                 {
+                    await _queueAutoResetEvent.WaitOneAsync(cancellationToken).ConfigureAwait(false);
                     return;
                 }
 
@@ -115,7 +133,7 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                await adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, packet, cancellationToken).ConfigureAwait(false);
+                adapter.SendPacketAsync(packet, cancellationToken).GetAwaiter().GetResult();
 
                 _logger.Verbose("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
             }

Source/MQTTnet/Server/MqttClientSession.cs

@@ -12,13 +12,13 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
-    public class MqttClientSession : IDisposable
+    public class MqttClientSession : IMqttClientSession
     {
         private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
 
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
         private readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
-        private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
+        private readonly MqttClientPendingPacketsQueue _pendingPacketsQueue;
         private readonly MqttClientSubscriptionsManager _subscriptionsManager;
         private readonly MqttClientSessionsManager _sessionsManager;
 
@@ -47,9 +47,9 @@ namespace MQTTnet.Server
 
             _logger = logger.CreateChildLogger(nameof(MqttClientSession));
 
-            _keepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, () => Stop(MqttClientDisconnectType.NotClean), _logger);
+            _keepAliveMonitor = new MqttClientKeepAliveMonitor(this, _logger);
             _subscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, sessionsManager.Server);
-            _pendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
+            _pendingPacketsQueue = new MqttClientPendingPacketsQueue(_options, this, _logger);
         }
 
         public string ClientId { get; }
@@ -60,7 +60,7 @@ namespace MQTTnet.Server
             status.IsConnected = _adapter != null;
             status.Endpoint = _adapter?.Endpoint;
             status.ProtocolVersion = _adapter?.PacketSerializer?.ProtocolVersion;
-            status.PendingApplicationMessagesCount = _pendingMessagesQueue.Count;
+            status.PendingApplicationMessagesCount = _pendingPacketsQueue.Count;
             status.LastPacketReceived = _keepAliveMonitor.LastPacketReceived;
             status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived;
         }
@@ -80,7 +80,7 @@ namespace MQTTnet.Server
                 _wasCleanDisconnect = false;
                 _willMessage = connectPacket.WillMessage;
 
-                _pendingMessagesQueue.Start(adapter, _cancellationTokenSource.Token);
+                _pendingPacketsQueue.Start(adapter, _cancellationTokenSource.Token);
                 _keepAliveMonitor.Start(connectPacket.KeepAlivePeriod, _cancellationTokenSource.Token);
 
                 while (!_cancellationTokenSource.IsCancellationRequested)
@@ -89,7 +89,7 @@ namespace MQTTnet.Server
                     if (packet != null)
                     {
                         _keepAliveMonitor.PacketReceived(packet);
-                        await ProcessReceivedPacketAsync(adapter, packet, _cancellationTokenSource.Token).ConfigureAwait(false);
+                        ProcessReceivedPacket(adapter, packet, _cancellationTokenSource.Token);
                     }
                 }
             }
@@ -102,7 +102,7 @@ namespace MQTTnet.Server
                 {
                     if (exception is MqttCommunicationClosedGracefullyException)
                     {
-                        _logger.Verbose("Client '{0}': Connection closed gracefully.", ClientId); ;
+                        _logger.Verbose("Client '{0}': Connection closed gracefully.", ClientId);
                     }
                     else
                     {
@@ -113,7 +113,7 @@ namespace MQTTnet.Server
                 {
                     _logger.Error(exception, "Client '{0}': Unhandled exception while receiving client packets.", ClientId);
                 }
-                
+
                 Stop(MqttClientDisconnectType.NotClean);
             }
             finally
@@ -123,7 +123,7 @@ namespace MQTTnet.Server
                     _adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
                     _adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
                 }
-                
+
                 _adapter = null;
 
                 _cancellationTokenSource?.Dispose();
@@ -149,13 +149,10 @@ namespace MQTTnet.Server
 
                 if (_willMessage != null && !_wasCleanDisconnect)
                 {
-                    _sessionsManager.StartDispatchApplicationMessage(this, _willMessage);
+                    _sessionsManager.EnqueueApplicationMessage(this, _willMessage.ToPublishPacket());
                 }
 
                 _willMessage = null;
-
-                ////_pendingMessagesQueue.WaitForCompletion();
-                ////_keepAliveMonitor.WaitForCompletion();
             }
             finally
             {
@@ -163,18 +160,24 @@ namespace MQTTnet.Server
             }
         }
 
-        public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
+        public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
         {
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
-            var checkSubscriptionsResult = _subscriptionsManager.CheckSubscriptions(applicationMessage);
+            var checkSubscriptionsResult = _subscriptionsManager.CheckSubscriptions(publishPacket.Topic, publishPacket.QualityOfServiceLevel);
             if (!checkSubscriptionsResult.IsSubscribed)
             {
                 return;
             }
 
-            var publishPacket = applicationMessage.ToPublishPacket();
-            publishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
+            publishPacket = new MqttPublishPacket
+            {
+                Topic = publishPacket.Topic,
+                Payload = publishPacket.Payload,
+                QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel,
+                Retain = false,
+                Dup = false
+            };
 
             if (publishPacket.QualityOfServiceLevel > 0)
             {
@@ -187,16 +190,20 @@ namespace MQTTnet.Server
                     senderClientSession?.ClientId,
                     ClientId,
                     publishPacket.ToApplicationMessage());
-               
+
                 _options.ClientMessageQueueInterceptor?.Invoke(context);
 
                 if (!context.AcceptEnqueue || context.ApplicationMessage == null)
                 {
                     return;
                 }
+
+                publishPacket.Topic = context.ApplicationMessage.Topic;
+                publishPacket.Payload = context.ApplicationMessage.Payload;
+                publishPacket.QualityOfServiceLevel = context.ApplicationMessage.QualityOfServiceLevel;
             }
-            
-            _pendingMessagesQueue.Enqueue(publishPacket);
+
+            _pendingPacketsQueue.Enqueue(publishPacket);
         }
 
         public Task SubscribeAsync(IList<TopicFilter> topicFilters)
@@ -226,31 +233,39 @@ namespace MQTTnet.Server
 
         public void ClearPendingApplicationMessages()
         {
-            _pendingMessagesQueue.Clear();
+            _pendingPacketsQueue.Clear();
         }
 
         public void Dispose()
         {
-            _pendingMessagesQueue?.Dispose();
+            _pendingPacketsQueue?.Dispose();
 
             _cancellationTokenSource?.Dispose();
         }
 
-        private Task ProcessReceivedPacketAsync(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken)
+        private void ProcessReceivedPacket(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken)
         {
             if (packet is MqttPublishPacket publishPacket)
             {
-                return HandleIncomingPublishPacketAsync(adapter, publishPacket, cancellationToken);
+                HandleIncomingPublishPacket(adapter, publishPacket, cancellationToken);
+                return;
             }
 
             if (packet is MqttPingReqPacket)
             {
-                return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, new MqttPingRespPacket(), cancellationToken);
+                adapter.SendPacketAsync(new MqttPingRespPacket(), cancellationToken).GetAwaiter().GetResult();
+                return;
             }
 
             if (packet is MqttPubRelPacket pubRelPacket)
             {
-                return HandleIncomingPubRelPacketAsync(adapter, pubRelPacket, cancellationToken);
+                var responsePacket = new MqttPubCompPacket
+                {
+                    PacketIdentifier = pubRelPacket.PacketIdentifier
+                };
+
+                adapter.SendPacketAsync(responsePacket, cancellationToken).GetAwaiter().GetResult();
+                return;
             }
 
             if (packet is MqttPubRecPacket pubRecPacket)
@@ -260,40 +275,41 @@ namespace MQTTnet.Server
                     PacketIdentifier = pubRecPacket.PacketIdentifier
                 };
 
-                return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, responsePacket, cancellationToken);
+                adapter.SendPacketAsync(responsePacket, cancellationToken).GetAwaiter().GetResult();
+                return;
             }
 
             if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
             {
-                // Discard message.
-                return Task.FromResult(0);
+                return;
             }
 
             if (packet is MqttSubscribePacket subscribePacket)
             {
-                return HandleIncomingSubscribePacketAsync(adapter, subscribePacket, cancellationToken);
+                HandleIncomingSubscribePacket(adapter, subscribePacket, cancellationToken);
+                return;
             }
 
             if (packet is MqttUnsubscribePacket unsubscribePacket)
             {
-                return HandleIncomingUnsubscribePacketAsync(adapter, unsubscribePacket, cancellationToken);
+                HandleIncomingUnsubscribePacket(adapter, unsubscribePacket, cancellationToken);
+                return;
             }
 
             if (packet is MqttDisconnectPacket)
             {
                 Stop(MqttClientDisconnectType.Clean);
-                return Task.FromResult(0);
+                return;
             }
 
             if (packet is MqttConnectPacket)
             {
                 Stop(MqttClientDisconnectType.NotClean);
-                return Task.FromResult(0);
+                return;
             }
 
             _logger.Warning(null, "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
             Stop(MqttClientDisconnectType.NotClean);
-            return Task.FromResult(0);
         }
 
         private void EnqueueSubscribedRetainedMessages(ICollection<TopicFilter> topicFilters)
@@ -301,14 +317,14 @@ namespace MQTTnet.Server
             var retainedMessages = _retainedMessagesManager.GetSubscribedMessages(topicFilters);
             foreach (var applicationMessage in retainedMessages)
             {
-                EnqueueApplicationMessage(null, applicationMessage);
+                EnqueueApplicationMessage(null, applicationMessage.ToPublishPacket());
             }
         }
 
-        private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
+        private void HandleIncomingSubscribePacket(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
         {
             var subscribeResult = _subscriptionsManager.Subscribe(subscribePacket);
-            await adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, subscribeResult.ResponsePacket, cancellationToken).ConfigureAwait(false);
+            adapter.SendPacketAsync(subscribeResult.ResponsePacket, cancellationToken).GetAwaiter().GetResult();
 
             if (subscribeResult.CloseConnection)
             {
@@ -319,30 +335,30 @@ namespace MQTTnet.Server
             EnqueueSubscribedRetainedMessages(subscribePacket.TopicFilters);
         }
 
-        private Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
+        private void HandleIncomingUnsubscribePacket(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
         {
             var unsubscribeResult = _subscriptionsManager.Unsubscribe(unsubscribePacket);
-            return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, unsubscribeResult, cancellationToken);
+            adapter.SendPacketAsync(unsubscribeResult, cancellationToken).GetAwaiter().GetResult();
         }
 
-        private Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        private void HandleIncomingPublishPacket(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
-            var applicationMessage = publishPacket.ToApplicationMessage();
-
-            switch (applicationMessage.QualityOfServiceLevel)
+            switch (publishPacket.QualityOfServiceLevel)
             {
                 case MqttQualityOfServiceLevel.AtMostOnce:
                     {
-                        _sessionsManager.StartDispatchApplicationMessage(this, applicationMessage);
-                        return Task.FromResult(0);
+                        HandleIncomingPublishPacketWithQoS0(publishPacket);
+                        break;
                     }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
                     {
-                        return HandleIncomingPublishPacketWithQoS1(adapter, applicationMessage, publishPacket, cancellationToken);
+                        HandleIncomingPublishPacketWithQoS1(adapter, publishPacket, cancellationToken);
+                        break;
                     }
                 case MqttQualityOfServiceLevel.ExactlyOnce:
                     {
-                        return HandleIncomingPublishPacketWithQoS2(adapter, applicationMessage, publishPacket, cancellationToken);
+                        HandleIncomingPublishPacketWithQoS2(adapter, publishPacket, cancellationToken);
+                        break;
                     }
                 default:
                     {
@@ -351,27 +367,40 @@ namespace MQTTnet.Server
             }
         }
 
-        private Task HandleIncomingPublishPacketWithQoS1(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        private void HandleIncomingPublishPacketWithQoS0(MqttPublishPacket publishPacket)
         {
-            _sessionsManager.StartDispatchApplicationMessage(this, applicationMessage);
-
-            var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier };
-            return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, response, cancellationToken);
+            _sessionsManager.EnqueueApplicationMessage(this, publishPacket);
         }
 
-        private Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        private void HandleIncomingPublishPacketWithQoS1(
+            IMqttChannelAdapter adapter,
+            MqttPublishPacket publishPacket,
+            CancellationToken cancellationToken)
+        {
+            _sessionsManager.EnqueueApplicationMessage(this, publishPacket);
+
+            var response = new MqttPubAckPacket
+            {
+                PacketIdentifier = publishPacket.PacketIdentifier
+            };
+
+            adapter.SendPacketAsync(response, cancellationToken).GetAwaiter().GetResult();
+        }
+
+        private void HandleIncomingPublishPacketWithQoS2(
+            IMqttChannelAdapter adapter,
+            MqttPublishPacket publishPacket,
+            CancellationToken cancellationToken)
         {
             // QoS 2 is implement as method "B" (4.3.3 QoS 2: Exactly once delivery)
-            _sessionsManager.StartDispatchApplicationMessage(this, applicationMessage);
+            _sessionsManager.EnqueueApplicationMessage(this, publishPacket);
 
-            var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier };
-            return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, response, cancellationToken);
-        }
+            var response = new MqttPubRecPacket
+            {
+                PacketIdentifier = publishPacket.PacketIdentifier
+            };
 
-        private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
-        {
-            var response = new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier };
-            return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, response, cancellationToken);
+            adapter.SendPacketAsync(response, cancellationToken).GetAwaiter().GetResult();
         }
 
         private void OnAdapterReadingPacketCompleted(object sender, EventArgs e)

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
@@ -14,19 +15,26 @@ namespace MQTTnet.Server
 {
     public class MqttClientSessionsManager : IDisposable
     {
-        private readonly ConcurrentDictionary<string, MqttClientSession> _sessions = new ConcurrentDictionary<string, MqttClientSession>();
-        private readonly AsyncLock _sessionPreparationLock = new AsyncLock();
+        private readonly BlockingCollection<MqttEnqueuedApplicationMessage> _messageQueue = new BlockingCollection<MqttEnqueuedApplicationMessage>();
+
+        /// <summary>
+        /// manual locking dictionaries is faster than using concurrent dictionary
+        /// </summary>
+        private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
+
+        private readonly CancellationToken _cancellationToken;
 
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
         private readonly IMqttServerOptions _options;
         private readonly IMqttNetChildLogger _logger;
 
-        public MqttClientSessionsManager(IMqttServerOptions options, MqttServer server, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetChildLogger logger)
+        public MqttClientSessionsManager(IMqttServerOptions options, MqttServer server, MqttRetainedMessagesManager retainedMessagesManager, CancellationToken cancellationToken, IMqttNetChildLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
             _logger = logger.CreateChildLogger(nameof(MqttClientSessionsManager));
 
+            _cancellationToken = cancellationToken;
             _options = options ?? throw new ArgumentNullException(nameof(options));
             Server = server ?? throw new ArgumentNullException(nameof(server));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
@@ -34,7 +42,154 @@ namespace MQTTnet.Server
 
         public MqttServer Server { get; }
 
-        public async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
+        public void Start()
+        {
+            Task.Factory.StartNew(() => ProcessQueuedApplicationMessages(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
+        }
+
+        public void Stop()
+        {
+            lock (_sessions)
+            {
+                foreach (var session in _sessions)
+                {
+                    session.Value.Stop(MqttClientDisconnectType.NotClean);
+                }
+
+                _sessions.Clear();
+            }
+        }
+
+        public Task StartSession(IMqttChannelAdapter clientAdapter)
+        {
+            return Task.Run(() => RunSession(clientAdapter, _cancellationToken), _cancellationToken);
+        }
+
+        public Task<IList<IMqttClientSessionStatus>> GetClientStatusAsync()
+        {
+            var result = new List<IMqttClientSessionStatus>();
+
+            foreach (var session in GetSessions())
+            {
+                var status = new MqttClientSessionStatus(this, session);
+                session.FillStatus(status);
+
+                result.Add(status);
+            }
+
+            return Task.FromResult((IList<IMqttClientSessionStatus>)result);
+        }
+
+        public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
+            _messageQueue.Add(new MqttEnqueuedApplicationMessage(senderClientSession, publishPacket), _cancellationToken);
+        }
+
+        public Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
+        {
+            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            lock (_sessions)
+            {
+                if (!_sessions.TryGetValue(clientId, out var session))
+                {
+                    throw new InvalidOperationException($"Client session '{clientId}' is unknown.");
+                }
+
+                return session.SubscribeAsync(topicFilters);
+            }
+        }
+
+        public Task UnsubscribeAsync(string clientId, IList<string> topicFilters)
+        {
+            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            lock (_sessions)
+            {
+                if (!_sessions.TryGetValue(clientId, out var session))
+                {
+                    throw new InvalidOperationException($"Client session '{clientId}' is unknown.");
+                }
+
+                return session.UnsubscribeAsync(topicFilters);
+            }
+        }
+
+        public void DeleteSession(string clientId)
+        {
+            lock (_sessions)
+            {
+                _sessions.Remove(clientId);
+            }
+            _logger.Verbose("Session for client '{0}' deleted.", clientId);
+        }
+
+        public void Dispose()
+        {
+            _messageQueue?.Dispose();
+        }
+
+        private void ProcessQueuedApplicationMessages(CancellationToken cancellationToken)
+        {
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                try
+                {
+                    var enqueuedApplicationMessage = _messageQueue.Take(cancellationToken);
+                    var sender = enqueuedApplicationMessage.Sender;
+                    var applicationMessage = enqueuedApplicationMessage.PublishPacket.ToApplicationMessage();
+
+                    var interceptorContext = InterceptApplicationMessage(sender, applicationMessage);
+                    if (interceptorContext != null)
+                    {
+                        if (interceptorContext.CloseConnection)
+                        {
+                            enqueuedApplicationMessage.Sender.Stop(MqttClientDisconnectType.NotClean);
+                        }
+
+                        if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
+                        {
+                            return;
+                        }
+
+                        applicationMessage = interceptorContext.ApplicationMessage;
+                    }
+
+                    Server.OnApplicationMessageReceived(sender?.ClientId, applicationMessage);
+
+                    if (applicationMessage.Retain)
+                    {
+                        _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).GetAwaiter().GetResult();
+                    }
+
+                    foreach (var clientSession in GetSessions())
+                    {
+                        clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage.ToPublishPacket());
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                }
+                catch (Exception exception)
+                {
+                    _logger.Error(exception, "Unhandled exception while processing queued application message.");
+                }
+            }
+        }
+
+        private List<MqttClientSession> GetSessions()
+        {
+            lock (_sessions)
+            {
+                return _sessions.Values.ToList();
+            }
+        }
+
+        private async Task RunSession(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
         {
             var clientId = string.Empty;
             var wasCleanDisconnect = false;
@@ -60,7 +215,7 @@ namespace MQTTnet.Server
                 var connectReturnCode = ValidateConnection(connectPacket);
                 if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
                 {
-                    await clientAdapter.SendPacketAsync(_options.DefaultCommunicationTimeout,
+                    await clientAdapter.SendPacketAsync(
                         new MqttConnAckPacket
                         {
                             ConnectReturnCode = connectReturnCode
@@ -70,15 +225,15 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                var result = await PrepareClientSessionAsync(connectPacket).ConfigureAwait(false);
+                var result = PrepareClientSession(connectPacket);
                 var clientSession = result.Session;
 
-                await clientAdapter.SendPacketAsync(_options.DefaultCommunicationTimeout,
+                await clientAdapter.SendPacketAsync(
                     new MqttConnAckPacket
                     {
                         ConnectReturnCode = connectReturnCode,
                         IsSessionPresent = result.IsExistingSession
-                    }, 
+                    },
                     cancellationToken).ConfigureAwait(false);
 
                 Server.OnClientConnected(clientId);
@@ -113,73 +268,6 @@ namespace MQTTnet.Server
             }
         }
 
-        public Task StopAsync()
-        {
-            foreach (var session in _sessions)
-            {
-                session.Value.Stop(MqttClientDisconnectType.NotClean);
-            }
-
-            _sessions.Clear();
-            return Task.FromResult(0);
-        }
-
-        public Task<IList<IMqttClientSessionStatus>> GetClientStatusAsync()
-        {
-            var result = new List<IMqttClientSessionStatus>();
-            foreach (var session in _sessions)
-            {
-                var status = new MqttClientSessionStatus(this, session.Value);
-                session.Value.FillStatus(status);
-
-                result.Add(status);
-            }
-
-            return Task.FromResult((IList<IMqttClientSessionStatus>)result);
-        }
-
-        public void StartDispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
-        {
-            Task.Run(() => DispatchApplicationMessageAsync(senderClientSession, applicationMessage));
-        }
-
-        public Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
-        {
-            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
-            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
-
-            if (!_sessions.TryGetValue(clientId, out var session))
-            {
-                throw new InvalidOperationException($"Client session '{clientId}' is unknown.");
-            }
-
-            return session.SubscribeAsync(topicFilters);
-        }
-
-        public Task UnsubscribeAsync(string clientId, IList<string> topicFilters)
-        {
-            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
-            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
-
-            if (!_sessions.TryGetValue(clientId, out var session))
-            {
-                throw new InvalidOperationException($"Client session '{clientId}' is unknown.");
-            }
-
-            return session.UnsubscribeAsync(topicFilters);
-        }
-
-        public void DeleteSession(string clientId)
-        {
-            _sessions.TryRemove(clientId, out _);
-            _logger.Verbose("Session for client '{0}' deleted.", clientId);
-        }
-
-        public void Dispose()
-        {
-            _sessionPreparationLock?.Dispose();
-        }
-
         private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
         {
             if (_options.ConnectionValidator == null)
@@ -197,16 +285,16 @@ namespace MQTTnet.Server
             return context.ReturnCode;
         }
 
-        private async Task<GetOrCreateClientSessionResult> PrepareClientSessionAsync(MqttConnectPacket connectPacket)
+        private PrepareClientSessionResult PrepareClientSession(MqttConnectPacket connectPacket)
         {
-            using (await _sessionPreparationLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
+            lock (_sessions)
             {
                 var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession);
                 if (isSessionPresent)
                 {
                     if (connectPacket.CleanSession)
                     {
-                        _sessions.TryRemove(connectPacket.ClientId, out _);
+                        _sessions.Remove(connectPacket.ClientId);
 
                         clientSession.Stop(MqttClientDisconnectType.Clean);
                         clientSession.Dispose();
@@ -231,60 +319,19 @@ namespace MQTTnet.Server
                     _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
                 }
 
-                return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
+                return new PrepareClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
             }
         }
 
-        private async Task DispatchApplicationMessageAsync(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
+        private MqttApplicationMessageInterceptorContext InterceptApplicationMessage(MqttClientSession sender, MqttApplicationMessage applicationMessage)
         {
-            try
-            {
-                var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage);
-                if (interceptorContext != null)
-                {
-                    if (interceptorContext.CloseConnection)
-                    {
-                        senderClientSession.Stop(MqttClientDisconnectType.NotClean);
-                    }
-
-                    if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
-                    {
-                        return;
-                    }
-
-                    applicationMessage = interceptorContext.ApplicationMessage;
-                }
-
-                Server.OnApplicationMessageReceived(senderClientSession?.ClientId, applicationMessage);
-
-                if (applicationMessage.Retain)
-                {
-                    await _retainedMessagesManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
-                }
-
-                foreach (var clientSession in _sessions.Values)
-                {
-                    clientSession.EnqueueApplicationMessage(senderClientSession, applicationMessage);
-                }
-            }
-            catch (Exception exception)
-            {
-                _logger.Error(exception, "Error while processing application message");
-            }
-        }
-
-        private MqttApplicationMessageInterceptorContext InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
-        {
-            var interceptorContext = new MqttApplicationMessageInterceptorContext(
-                senderClientSession?.ClientId,
-                applicationMessage);
-
             var interceptor = _options.ApplicationMessageInterceptor;
             if (interceptor == null)
             {
-                return interceptorContext;
+                return null;
             }
 
+            var interceptorContext = new MqttApplicationMessageInterceptorContext(sender?.ClientId, applicationMessage);
             interceptor(interceptorContext);
             return interceptorContext;
         }

Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using MQTTnet.Packets;
@@ -9,7 +8,7 @@ namespace MQTTnet.Server
 {
     public class MqttClientSubscriptionsManager
     {
-        private readonly ConcurrentDictionary<string, MqttQualityOfServiceLevel> _subscriptions = new ConcurrentDictionary<string, MqttQualityOfServiceLevel>();
+        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         private readonly IMqttServerOptions _options;
         private readonly MqttServer _server;
         private readonly string _clientId;
@@ -54,7 +53,11 @@ namespace MQTTnet.Server
 
                 if (interceptorContext.AcceptSubscription)
                 {
-                    _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                    lock (_subscriptions)
+                    {
+                        _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                    }
+
                     _server.OnClientSubscribedTopic(_clientId, topicFilter);
                 }
             }
@@ -66,10 +69,14 @@ namespace MQTTnet.Server
         {
             if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
 
-            foreach (var topicFilter in unsubscribePacket.TopicFilters)
+            lock (_subscriptions)
             {
-                _subscriptions.TryRemove(topicFilter, out _);
-                _server.OnClientUnsubscribedTopic(_clientId, topicFilter);
+                foreach (var topicFilter in unsubscribePacket.TopicFilters)
+                {
+                    _subscriptions.Remove(topicFilter);
+
+                    _server.OnClientUnsubscribedTopic(_clientId, topicFilter);
+                }
             }
 
             return new MqttUnsubAckPacket
@@ -78,19 +85,21 @@ namespace MQTTnet.Server
             };
         }
 
-        public CheckSubscriptionsResult CheckSubscriptions(MqttApplicationMessage applicationMessage)
+        public CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel qosLevel)
         {
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
-
             var qosLevels = new HashSet<MqttQualityOfServiceLevel>();
-            foreach (var subscription in _subscriptions)
-            {
-                if (!MqttTopicFilterComparer.IsMatch(applicationMessage.Topic, subscription.Key))
-                {
-                    continue;
-                }
 
-                qosLevels.Add(subscription.Value);
+            lock (_subscriptions)
+            {
+                foreach (var subscription in _subscriptions)
+                {
+                    if (!MqttTopicFilterComparer.IsMatch(topic, subscription.Key))
+                    {
+                        continue;
+                    }
+
+                    qosLevels.Add(subscription.Value);
+                }
             }
 
             if (qosLevels.Count == 0)
@@ -101,7 +110,7 @@ namespace MQTTnet.Server
                 };
             }
 
-            return CreateSubscriptionResult(applicationMessage, qosLevels);
+            return CreateSubscriptionResult(qosLevel, qosLevels);
         }
 
         private static MqttSubscribeReturnCode ConvertToMaximumQoS(MqttQualityOfServiceLevel qualityOfServiceLevel)
@@ -122,12 +131,12 @@ namespace MQTTnet.Server
             return interceptorContext;
         }
 
-        private static CheckSubscriptionsResult CreateSubscriptionResult(MqttApplicationMessage applicationMessage, HashSet<MqttQualityOfServiceLevel> subscribedQoSLevels)
+        private static CheckSubscriptionsResult CreateSubscriptionResult(MqttQualityOfServiceLevel qosLevel, HashSet<MqttQualityOfServiceLevel> subscribedQoSLevels)
         {
             MqttQualityOfServiceLevel effectiveQoS;
-            if (subscribedQoSLevels.Contains(applicationMessage.QualityOfServiceLevel))
+            if (subscribedQoSLevels.Contains(qosLevel))
             {
-                effectiveQoS = applicationMessage.QualityOfServiceLevel;
+                effectiveQoS = qosLevel;
             }
             else if (subscribedQoSLevels.Count == 1)
             {

Source/MQTTnet/Server/MqttEnqueuedApplicationMessage.cs

@@ -0,0 +1,17 @@
+using MQTTnet.Packets;
+
+namespace MQTTnet.Server
+{
+    public class MqttEnqueuedApplicationMessage
+    {
+        public MqttEnqueuedApplicationMessage(MqttClientSession sender, MqttPublishPacket publishPacket)
+        {
+            Sender = sender;
+            PublishPacket = publishPacket;
+        }
+
+        public MqttClientSession Sender { get; }
+
+        public MqttPublishPacket PublishPacket { get; }
+    }
+}

Source/MQTTnet/Server/MqttRetainedMessagesManager.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
@@ -9,7 +8,8 @@ namespace MQTTnet.Server
 {
     public class MqttRetainedMessagesManager
     {
-        private readonly ConcurrentDictionary<string, MqttApplicationMessage> _messages = new ConcurrentDictionary<string, MqttApplicationMessage>();
+        private readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>();
+
         private readonly IMqttNetChildLogger _logger;
         private readonly IMqttServerOptions _options;
 
@@ -31,10 +31,13 @@ namespace MQTTnet.Server
             {
                 var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false);
 
-                _messages.Clear();
-                foreach (var retainedMessage in retainedMessages)
+                lock (_messages)
                 {
-                    _messages[retainedMessage.Topic] = retainedMessage;
+                    _messages.Clear();
+                    foreach (var retainedMessage in retainedMessages)
+                    {
+                        _messages[retainedMessage.Topic] = retainedMessage;
+                    }
                 }
             }
             catch (Exception exception)
@@ -61,17 +64,20 @@ namespace MQTTnet.Server
         {
             var retainedMessages = new List<MqttApplicationMessage>();
 
-            foreach (var retainedMessage in _messages.Values)
+            lock (_messages)
             {
-                foreach (var topicFilter in topicFilters)
+                foreach (var retainedMessage in _messages.Values)
                 {
-                    if (!MqttTopicFilterComparer.IsMatch(retainedMessage.Topic, topicFilter.Topic))
+                    foreach (var topicFilter in topicFilters)
                     {
-                        continue;
-                    }
+                        if (!MqttTopicFilterComparer.IsMatch(retainedMessage.Topic, topicFilter.Topic))
+                        {
+                            continue;
+                        }
 
-                    retainedMessages.Add(retainedMessage);
-                    break;
+                        retainedMessages.Add(retainedMessage);
+                        break;
+                    }
                 }
             }
 
@@ -82,28 +88,31 @@ namespace MQTTnet.Server
         {
             var saveIsRequired = false;
 
-            if (applicationMessage.Payload?.Length == 0)
+            lock (_messages)
             {
-                saveIsRequired = _messages.TryRemove(applicationMessage.Topic, out _);
-                _logger.Info("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
-            }
-            else
-            {
-                if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage))
+                if (applicationMessage.Payload?.Length == 0)
                 {
-                    _messages[applicationMessage.Topic] = applicationMessage;
-                    saveIsRequired = true;
+                    saveIsRequired = _messages.Remove(applicationMessage.Topic);
+                    _logger.Info("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
                 }
                 else
                 {
-                    if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
+                    if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage))
                     {
                         _messages[applicationMessage.Topic] = applicationMessage;
                         saveIsRequired = true;
                     }
-                }
+                    else
+                    {
+                        if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
+                        {
+                            _messages[applicationMessage.Topic] = applicationMessage;
+                            saveIsRequired = true;
+                        }
+                    }
 
-                _logger.Info("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
+                    _logger.Info("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
+                }
             }
 
             if (!saveIsRequired)
@@ -113,7 +122,13 @@ namespace MQTTnet.Server
 
             if (saveIsRequired && _options.Storage != null)
             {
-                await _options.Storage.SaveRetainedMessagesAsync(_messages.Values.ToList()).ConfigureAwait(false);
+                List<MqttApplicationMessage> messages;
+                lock (_messages)
+                {
+                    messages = _messages.Values.ToList();
+                }
+
+                await _options.Storage.SaveRetainedMessagesAsync(messages).ConfigureAwait(false);
             }
         }
     }

Source/MQTTnet/Server/MqttServer.cs

@@ -5,6 +5,7 @@ using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
+using MQTTnet.Internal;
 
 namespace MQTTnet.Server
 {
@@ -65,7 +66,7 @@ namespace MQTTnet.Server
 
             if (_cancellationTokenSource == null) throw new InvalidOperationException("The server is not started.");
 
-            _clientSessionsManager.StartDispatchApplicationMessage(null, applicationMessage);
+            _clientSessionsManager.EnqueueApplicationMessage(null, applicationMessage.ToPublishPacket());
 
             return Task.FromResult(0);
         }
@@ -81,7 +82,8 @@ namespace MQTTnet.Server
             _retainedMessagesManager = new MqttRetainedMessagesManager(Options, _logger);
             await _retainedMessagesManager.LoadMessagesAsync().ConfigureAwait(false);
 
-            _clientSessionsManager = new MqttClientSessionsManager(Options, this, _retainedMessagesManager, _logger);
+            _clientSessionsManager = new MqttClientSessionsManager(Options, this, _retainedMessagesManager, _cancellationTokenSource.Token, _logger);
+            _clientSessionsManager.Start();
 
             foreach (var adapter in _adapters)
             {
@@ -103,25 +105,26 @@ namespace MQTTnet.Server
                 }
 
                 _cancellationTokenSource.Cancel(false);
-                _cancellationTokenSource.Dispose();
-
+                
                 foreach (var adapter in _adapters)
                 {
                     adapter.ClientAccepted -= OnClientAccepted;
                     await adapter.StopAsync().ConfigureAwait(false);
                 }
 
-                await _clientSessionsManager.StopAsync().ConfigureAwait(false);
+                _clientSessionsManager.Stop();
 
                 _logger.Info("Stopped.");
                 Stopped?.Invoke(this, EventArgs.Empty);
             }
             finally
             {
-                _clientSessionsManager?.Dispose();
-                
+                _cancellationTokenSource?.Dispose();
                 _cancellationTokenSource = null;
+
                 _retainedMessagesManager = null;
+
+                _clientSessionsManager?.Dispose();
                 _clientSessionsManager = null;
             }
         }
@@ -155,9 +158,7 @@ namespace MQTTnet.Server
 
         private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
         {
-            eventArgs.SessionTask = Task.Run(
-                () => _clientSessionsManager.RunSessionAsync(eventArgs.Client, _cancellationTokenSource.Token),
-                _cancellationTokenSource.Token);
+            eventArgs.SessionTask = _clientSessionsManager.StartSession(eventArgs.Client);
         }
     }
 }

Source/MQTTnet/Server/PrepareClientSessionResult.cs

@@ -1,6 +1,6 @@
 namespace MQTTnet.Server
 {
-    public class GetOrCreateClientSessionResult
+    public class PrepareClientSessionResult
     {
         public bool IsExistingSession { get; set; }