Update project files.

Based on https://github.com/chkr1011/MQTTnet/issues/1138#issuecomment-846649937

.github/workflows/deb-packaging.yml

@@ -0,0 +1,62 @@
+# This is a basic workflow to help you get started with Actions
+
+name: Deb Packaging
+
+# Controls when the action will run. 
+on:
+  # Allows you to run this workflow manually from the Actions tab
+  workflow_dispatch
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Parse version
+        run: echo VERSION=$(grep '<Version>' Source/MQTTnet.Server/MQTTnet.Server.csproj | grep [0-9\.]* -o) >> $GITHUB_ENV
+
+      - name: Build MQTTnetServer
+        run: dotnet publish Source/MQTTnet.Server/MQTTnet.Server.csproj --configuration Release --self-contained --runtime linux-x64 --framework net5.0
+
+      - name: Set up installation directory
+        run: mkdir -p packaging/opt/MQTTnetServer
+
+      - name: Copy meta data
+        run: |
+          cp -R Build/deb-meta/* packaging/
+          chmod 755 packaging/DEBIAN/postinst
+          chmod 755 packaging/DEBIAN/postrm
+          chmod 755 packaging/DEBIAN/prerm
+
+      - name: Move artifacts to packaging directory
+        run: cp -R Source/MQTTnet.Server/bin/Release/net5.0/linux-x64/publish/* packaging/opt/MQTTnetServer
+
+      - name: Adjust files
+        run: |
+          rm packaging/opt/MQTTnetServer/appsettings.Development.json
+          mv packaging/opt/MQTTnetServer/appsettings.json packaging/opt/MQTTnetServer/appsettings.template.json
+
+      - name: Adjust permissions
+        run: |
+          cd packaging/opt/MQTTnetServer
+          find . -type f | xargs chmod -R 644
+          chmod 755 MQTTnet.Server
+
+      - name: Generate MD5s
+        run: |
+          cd packaging/          
+          md5sum $(find * -type f -not -path 'DEBIAN/*') > DEBIAN/md5sums
+      
+      - name: Patch meta
+        run: sed -i 's/\VERSIONPLACEHOLDER/${{ env.VERSION }}/' packaging/DEBIAN/control
+
+      - name: Package everything
+        run: dpkg-deb -v --build packaging/ mqttnet-server_${{ env.VERSION }}-1_amd64.deb
+      
+      - name: Save artifact
+        uses: actions/upload-artifact@v2
+        with:
+          name: mqttnet-server
+          path: mqttnet-server_${{ env.VERSION }}-1_amd64.deb

.gitignore

@@ -294,3 +294,5 @@ __pycache__/
 /Tests/MQTTnet.TestApp.NetCore/RetainedMessages.json
 
 Build/NuGet/
+
+Build/nuget.exe

Build/MQTTnet.nuspec

@@ -12,13 +12,14 @@
     <requireLicenseAcceptance>true</requireLicenseAcceptance>
     <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</description>
     <releaseNotes>
-* [Core] Fixed some issues in nuget packages.
+* [Core] Fixed _DISCONNECT_ packet reading for protocol version 3.1.0.DisconnectPacket
-* [Client] Changed exception types so that proper exceptions are thrown when connecting (#1082).
+* [Client] Added support for deferred message approval (#1075, thanks to @tkurucsai).
-* [Client] Exposed Dup flag in application messages.
+* [Client] Exposed missing APIs for .NET 5.0 build (thanks to @yyjdelete).
-* [Client] Improved keep alive message sending.
+* [Client] Improved disconnect handling (#1134, thanks to @yyjdelete).
-* [RpcClient] Fixed an issue when using a custom application message reveived handler (#1093).
+* [Client] Fix TLS parameter options builder (#1104).
-* [RpcClient] Fixed a race condition when using timeout and cancellation token at the same time (BREAKING CHANGE!).
+* [Server] Add new overload for options builder.
-* [Server] Improved keep alive message sending.
+* [Server] Exposed endpoint in _MqttServerClientDisconnectedEventArgs_.
+* [Server] Remove client disconnected handler from wrong implementation location (BREAKING CHANGE).
 
 Git commit: $gitCommit
     </releaseNotes>

Build/deb-meta/DEBIAN/control

@@ -0,0 +1,10 @@
+Package: mqttnet-server
+Version: VERSIONPLACEHOLDER
+Section: base
+Priority: optional
+Architecture: amd64
+Depends: libssl1.1
+Suggests:
+Maintainer: Damian Wolgast <packages@asymetrixs.net>
+Description:
+  MQTTnet Server is a standalone cross platform MQTT server (like mosquitto) basing on this library.

Build/deb-meta/DEBIAN/copyright

@@ -0,0 +1,26 @@
+The source of MQTTnet is at:
+https://github.com/chkr1011/MQTTnet
+Project is maintained by Christian Kratky
+
+
+MIT License
+
+Copyright (c) 2021 Damian Wolgast
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Build/deb-meta/DEBIAN/postinst

@@ -0,0 +1,44 @@
+#!/bin/bash
+
+# Check if user exits
+USER=mqttnetsrv
+USEREXISTS=$(id -u $USER >/dev/null 2>&1; echo $?)
+EC=0
+BASEPATH=/opt
+INSTALLPATH=$BASEPATH/MQTTnetServer
+
+# If user does not exist, create it
+if (( $USEREXISTS == 1 )); then
+    useradd -d $INSTALLPATH  -r $USER
+    EC=$(echo $?)
+
+    if (( $EC == 1 )); then
+        exit 1;
+    fi
+fi
+
+# Set permissions on files
+chown -R root:root $INSTALLPATH
+
+# Check if config does not exist and create it
+if [ ! -f "$INSTALLPATH/appsettings.json" ]; then
+    cp $INSTALLPATH/appsettings.template.json $INSTALLPATH/appsettings.json
+fi
+
+
+# Check if setcap exists, does not exist in docker containers
+SCAP=$(which setcap >/dev/null; echo $?)
+if [ "$SCAP" -eq "0" ]; then
+    # Enable MQTTServer.NET to bind to IP interface
+    setcap CAP_NET_BIND_SERVICE=+eip $INSTALLPATH/MQTTnet.Server
+fi
+
+chmod 644 /etc/systemd/system/mqttnet-server.service
+
+# Check if systemctl exists, does not exist in docker containers
+SCTL=$(which systemctl >/dev/null; echo $?)
+
+if [ "$SCTL" -eq "0" ]; then
+    # Reload systemd because of new service file
+    systemctl daemon-reload
+fi

Build/deb-meta/DEBIAN/postrm

@@ -0,0 +1,8 @@
+#!/bin/bash
+
+# Check if systemctl exists, does not exist in docker containers
+SCTL=$(which systemctl >/dev/null; echo $?)
+if [ "$SCTL" -eq "0" ]; then
+    # Reload systemd because of new service file
+    systemctl daemon-reload
+fi

Build/deb-meta/DEBIAN/prerm

@@ -0,0 +1,18 @@
+#!/bin/bash
+
+BASEPATH=/opt
+INSTALLPATH=$BASEPATH/MQTTnetServer
+
+# Check if systemctl exists, does not exist in docker containers
+SCTL=$(which systemctl >/dev/null; echo $?)
+if [ "$SCTL" -eq "0" ]; then
+    # Stop service before uninstalling
+    systemctl stop mqttnet-server
+fi
+
+# Check if setcap exists, does not exist in docker containers
+SCAP=$(which setcap >/dev/null; echo $?)
+if [ "$SCAP" -eq "0" ]; then
+    # Remove permission to open ports
+    setcap CAP_NET_BIND_SERVICE=-eip $INSTALLPATH/MQTTnet.Server
+fi

Build/deb-meta/etc/systemd/system/mqttnet-server.service

@@ -0,0 +1,14 @@
+[Unit]
+Description=MQTTnetServer
+After=network.target
+
+[Service]
+ExecStart=/opt/MQTTnetServer/MQTTnet.Server
+WorkingDirectory=/opt/MQTTnetServer
+StandardOutput=inherit
+StandardError=inherit
+Restart=always
+User=mqttnetsrv
+
+[Install]
+WantedBy=multi-user.target

Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;net5.0</TargetFrameworks>
@@ -8,6 +8,9 @@
     <Company />
     <Authors />
     <PackageId />
+    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
+    <SignAssembly>false</SignAssembly>
+    <DelaySign>false</DelaySign>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
     <IncludeSymbols>true</IncludeSymbols>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>

Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;net5.0</TargetFrameworks>
@@ -8,6 +8,9 @@
     <Company />
     <Authors />
     <PackageId />
+    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
+    <SignAssembly>false</SignAssembly>
+    <DelaySign>false</DelaySign>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
     <IncludeSymbols>true</IncludeSymbols>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>

Source/MQTTnet.Extensions.WebSocket4Net/MQTTnet.Extensions.WebSocket4Net.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;net5.0</TargetFrameworks>
@@ -8,6 +8,9 @@
     <Company />
     <Authors />
     <PackageId />
+    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
+    <SignAssembly>false</SignAssembly>
+    <DelaySign>false</DelaySign>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
     <IncludeSymbols>true</IncludeSymbols>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>

Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs

@@ -55,7 +55,7 @@ namespace MQTTnet.Server.Logging
                 case MqttNetLogLevel.Error: return LogLevel.Error;
                 case MqttNetLogLevel.Warning: return LogLevel.Warning;
                 case MqttNetLogLevel.Info: return LogLevel.Information;
-                case MqttNetLogLevel.Verbose: return LogLevel.Debug;
+                case MqttNetLogLevel.Verbose: return LogLevel.Trace;
             }
 
             return LogLevel.Debug;

Source/MQTTnet/Client/MqttClient.cs

@@ -38,10 +38,10 @@ namespace MQTTnet.Client
 
         IMqttChannelAdapter _adapter;
         bool _cleanDisconnectInitiated;
-        long _isDisconnectPending;
+        volatile int _connectionStatus;
-        bool _isConnected;
+
         MqttClientDisconnectReason _disconnectReason;
-        
+
         DateTime _lastPacketSentTimestamp;
 
         public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
@@ -58,7 +58,7 @@ namespace MQTTnet.Client
 
         public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; }
 
-        public bool IsConnected => _isConnected && Interlocked.Read(ref _isDisconnectPending) == 0;
+        public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected;
 
         public IMqttClientOptions Options { get; private set; }
 
@@ -71,6 +71,11 @@ namespace MQTTnet.Client
 
             ThrowIfDisposed();
 
+            if (CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connecting, MqttClientConnectionStatus.Disconnected) != MqttClientConnectionStatus.Disconnected)
+            {
+                throw new InvalidOperationException("Not allowed to connect while connect/disconnect is pending.");
+            }
+
             MqttClientAuthenticateResult authenticateResult = null;
 
             try
@@ -83,7 +88,6 @@ namespace MQTTnet.Client
                 _backgroundCancellationTokenSource = new CancellationTokenSource();
                 var backgroundCancellationToken = _backgroundCancellationTokenSource.Token;
 
-                _isDisconnectPending = 0;
                 var adapter = _adapterFactory.CreateClientAdapter(options);
                 _adapter = adapter;
 
@@ -108,7 +112,7 @@ namespace MQTTnet.Client
                     _keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessagesAsync(backgroundCancellationToken), backgroundCancellationToken);
                 }
 
-                _isConnected = true;
+                CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connected, MqttClientConnectionStatus.Connecting);
 
                 _logger.Info("Connected.");
 
@@ -126,10 +130,7 @@ namespace MQTTnet.Client
 
                 _logger.Error(exception, "Error while connecting with server.");
 
-                if (!DisconnectIsPending())
+                await DisconnectInternalAsync(null, exception, authenticateResult).ConfigureAwait(false);
-                {
-                    await DisconnectInternalAsync(null, exception, authenticateResult).ConfigureAwait(false);
-                }
 
                 throw;
             }
@@ -141,7 +142,8 @@ namespace MQTTnet.Client
 
             ThrowIfDisposed();
 
-            if (DisconnectIsPending())
+            var clientWasConnected = IsConnected;
+            if (DisconnectIsPendingOrFinished())
             {
                 return;
             }
@@ -151,7 +153,7 @@ namespace MQTTnet.Client
                 _disconnectReason = MqttClientDisconnectReason.NormalDisconnection;
                 _cleanDisconnectInitiated = true;
 
-                if (_isConnected)
+                if (clientWasConnected)
                 {
                     var disconnectPacket = _adapter.PacketFormatterAdapter.DataConverter.CreateDisconnectPacket(options);
                     await SendAsync(disconnectPacket, cancellationToken).ConfigureAwait(false);
@@ -159,7 +161,7 @@ namespace MQTTnet.Client
             }
             finally
             {
-                await DisconnectInternalAsync(null, null, null).ConfigureAwait(false);
+                await DisconnectCoreAsync(null, null, null, clientWasConnected).ConfigureAwait(false);
             }
         }
 
@@ -230,21 +232,21 @@ namespace MQTTnet.Client
             switch (applicationMessage.QualityOfServiceLevel)
             {
                 case MqttQualityOfServiceLevel.AtMostOnce:
-                    {
+                {
-                        return PublishAtMostOnce(publishPacket, cancellationToken);
+                    return PublishAtMostOnce(publishPacket, cancellationToken);
-                    }
+                }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
-                    {
+                {
-                        return PublishAtLeastOnceAsync(publishPacket, cancellationToken);
+                    return PublishAtLeastOnceAsync(publishPacket, cancellationToken);
-                    }
+                }
                 case MqttQualityOfServiceLevel.ExactlyOnce:
-                    {
+                {
-                        return PublishExactlyOnceAsync(publishPacket, cancellationToken);
+                    return PublishExactlyOnceAsync(publishPacket, cancellationToken);
-                    }
+                }
                 default:
-                    {
+                {
-                        throw new NotSupportedException();
+                    throw new NotSupportedException();
-                    }
+                }
             }
         }
 
@@ -306,7 +308,7 @@ namespace MQTTnet.Client
 
         void ThrowIfNotConnected()
         {
-            if (!IsConnected || Interlocked.Read(ref _isDisconnectPending) == 1)
+            if (!IsConnected)
             {
                 throw new MqttCommunicationException("The client is not connected.");
             }
@@ -317,12 +319,21 @@ namespace MQTTnet.Client
             if (IsConnected) throw new MqttProtocolViolationException(message);
         }
 
-        async Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientAuthenticateResult authenticateResult)
+        Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientAuthenticateResult authenticateResult)
         {
-            var clientWasConnected = _isConnected;
+            var clientWasConnected = IsConnected;
+            
+            if (!DisconnectIsPendingOrFinished())
+            {
+                return DisconnectCoreAsync(sender, exception, authenticateResult, clientWasConnected);
+            }
+            
+            return PlatformAbstractionLayer.CompletedTask;
+        }
 
+        async Task DisconnectCoreAsync(Task sender, Exception exception, MqttClientAuthenticateResult authenticateResult, bool clientWasConnected)
+        {
             TryInitiateDisconnect();
-            _isConnected = false;
 
             try
             {
@@ -346,8 +357,6 @@ namespace MQTTnet.Client
                 var keepAliveTask = WaitForTaskAsync(_keepAlivePacketsSenderTask, sender);
 
                 await Task.WhenAll(receiverTask, publishPacketReceiverTask, keepAliveTask).ConfigureAwait(false);
-
-                _publishPacketReceiverQueue?.Dispose();
             }
             catch (Exception e)
             {
@@ -357,6 +366,7 @@ namespace MQTTnet.Client
             {
                 Cleanup();
                 _cleanDisconnectInitiated = false;
+                CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnected, MqttClientConnectionStatus.Disconnecting);
 
                 _logger.Info("Disconnected.");
 
@@ -390,7 +400,7 @@ namespace MQTTnet.Client
             cancellationToken.ThrowIfCancellationRequested();
 
             _lastPacketSentTimestamp = DateTime.UtcNow;
-            
+
             return _adapter.SendPacketAsync(packet, cancellationToken);
         }
 
@@ -477,10 +487,7 @@ namespace MQTTnet.Client
                     _logger.Error(exception, "Error exception while sending/receiving keep alive packets.");
                 }
 
-                if (!DisconnectIsPending())
+                await DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
-                {
-                    await DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
-                }
             }
             finally
             {
@@ -505,10 +512,7 @@ namespace MQTTnet.Client
 
                     if (packet == null)
                     {
-                        if (!DisconnectIsPending())
+                        await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false);
-                        {
-                            await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false);
-                        }
 
                         return;
                     }
@@ -537,10 +541,7 @@ namespace MQTTnet.Client
 
                 _packetDispatcher.FailAll(exception);
 
-                if (!DisconnectIsPending())
+                await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
-                {
-                    await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
-                }
             }
             finally
             {
@@ -609,10 +610,7 @@ namespace MQTTnet.Client
 
                 _packetDispatcher.FailAll(exception);
 
-                if (!DisconnectIsPending())
+                await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
-                {
-                    await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
-                }
             }
         }
 
@@ -641,34 +639,11 @@ namespace MQTTnet.Client
                     }
 
                     var publishPacket = publishPacketDequeueResult.Item;
+                    var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
 
-                    if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
+                    if (eventArgs.AutoAcknowledge)
                     {
-                        await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
+                        await eventArgs.AcknowledgeAsync(cancellationToken).ConfigureAwait(false);
-                    }
-                    else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
-                    {
-                        var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
-
-                        if (!eventArgs.ProcessingFailed)
-                        {
-                            var pubAckPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(publishPacket, eventArgs.ReasonCode);
-                            await SendAsync(pubAckPacket, cancellationToken).ConfigureAwait(false);
-                        }
-                    }
-                    else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
-                    {
-                        var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
-
-                        if (!eventArgs.ProcessingFailed)
-                        {
-                            var pubRecPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(publishPacket, eventArgs.ReasonCode);
-                            await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
-                        }
-                    }
-                    else
-                    {
-                        throw new MqttProtocolViolationException("Received a not supported QoS level.");
                     }
                 }
                 catch (OperationCanceledException)
@@ -681,6 +656,36 @@ namespace MQTTnet.Client
             }
         }
 
+        internal Task AcknowledgeReceivedPublishPacket(MqttApplicationMessageReceivedEventArgs eventArgs, CancellationToken cancellationToken)
+        {
+            if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
+            {
+                // no response required
+            }
+            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
+            {
+                if (!eventArgs.ProcessingFailed)
+                {
+                    var pubAckPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(eventArgs.PublishPacket, eventArgs.ReasonCode);
+                    return SendAsync(pubAckPacket, cancellationToken);
+                }
+            }
+            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
+            {
+                if (!eventArgs.ProcessingFailed)
+                {
+                    var pubRecPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(eventArgs.PublishPacket, eventArgs.ReasonCode);
+                    return SendAsync(pubRecPacket, cancellationToken);
+                }
+            }
+            else
+            {
+                throw new MqttProtocolViolationException("Received a not supported QoS level.");
+            }
+
+            return PlatformAbstractionLayer.CompletedTask;
+        }
+
         Task ProcessReceivedPubRecPacket(MqttPubRecPacket pubRecPacket, CancellationToken cancellationToken)
         {
             if (!_packetDispatcher.TryDispatch(pubRecPacket))
@@ -702,17 +707,12 @@ namespace MQTTnet.Client
 
         Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket)
         {
-            _disconnectReason = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.NormalDisconnection);
+            _disconnectReason = (MqttClientDisconnectReason) (disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.NormalDisconnection);
 
             // Also dispatch disconnect to waiting threads to generate a proper exception.
             _packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
-            
-            if (!DisconnectIsPending())
-            {
-                return DisconnectInternalAsync(_packetReceiverTask, null, null);
-            }
 
-            return PlatformAbstractionLayer.CompletedTask;
+            return DisconnectInternalAsync(_packetReceiverTask, null, null);
         }
 
         Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket)
@@ -736,8 +736,10 @@ namespace MQTTnet.Client
         async Task<MqttClientPublishResult> PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
             publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
-            var response = await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
+            
-            return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(response);
+            var pubAckPacket = await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
+            
+            return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(pubAckPacket);
         }
 
         async Task<MqttClientPublishResult> PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
@@ -747,7 +749,7 @@ namespace MQTTnet.Client
             var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
 
             var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
-            
+
             var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
 
             return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(pubRecPacket, pubCompPacket);
@@ -756,7 +758,7 @@ namespace MQTTnet.Client
         async Task<MqttApplicationMessageReceivedEventArgs> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
         {
             var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket);
-            var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage);
+            var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket);
 
             var handler = ApplicationMessageReceivedHandler;
             if (handler != null)
@@ -798,11 +800,36 @@ namespace MQTTnet.Client
             }
         }
 
-        bool DisconnectIsPending()
+        bool DisconnectIsPendingOrFinished()
         {
-            // This will read the _isDisconnectPending and set it to "1" afterwards regardless of the value.
+            var connectionStatus = (MqttClientConnectionStatus)_connectionStatus;
-            // So the first caller will get a "false" and all subsequent ones will get "true".
+            
-            return Interlocked.CompareExchange(ref _isDisconnectPending, 1, 0) != 0;
+            do
+            {
+                switch (connectionStatus)
+                {
+                    case MqttClientConnectionStatus.Disconnected:
+                    case MqttClientConnectionStatus.Disconnecting:
+                        return true;
+                    case MqttClientConnectionStatus.Connected:
+                    case MqttClientConnectionStatus.Connecting:
+                        // This will compare the _connectionStatus to old value and set it to "MqttClientConnectionStatus.Disconnecting" afterwards.
+                        // So the first caller will get a "false" and all subsequent ones will get "true".
+                        var curStatus = CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnecting, connectionStatus);
+                        if (curStatus == connectionStatus)
+                        {
+                            return false;
+                        }
+                        
+                        connectionStatus = curStatus;
+                        break;
+                }
+            } while (true);
+        }
+
+        MqttClientConnectionStatus CompareExchangeConnectionStatus(MqttClientConnectionStatus value, MqttClientConnectionStatus comparand)
+        {
+            return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand);
         }
     }
-}
+}

Source/MQTTnet/Client/MqttClientConnectionStatus.cs

@@ -0,0 +1,10 @@
+namespace MQTTnet.Client
+{
+    public enum MqttClientConnectionStatus
+    {
+        Disconnected,
+        Disconnecting,
+        Connected,
+        Connecting
+    }
+}

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -252,7 +252,11 @@ namespace MQTTnet.Client.Options
         {
             if (optionsBuilder == null) throw new ArgumentNullException(nameof(optionsBuilder));
 
-            _tlsParameters = new MqttClientOptionsBuilderTlsParameters();
+            _tlsParameters = new MqttClientOptionsBuilderTlsParameters
+            {
+                UseTls = true
+            };
+            
             optionsBuilder(_tlsParameters);
             return this;
         }
@@ -287,7 +291,7 @@ namespace MQTTnet.Client.Options
 #pragma warning disable CS0618 // Type or member is obsolete
                         CertificateValidationCallback = _tlsParameters.CertificateValidationCallback,
 #pragma warning restore CS0618 // Type or member is obsolete
-#if NETCOREAPP3_1
+#if NETCOREAPP3_1 || NET5_0
                         ApplicationProtocols = _tlsParameters.ApplicationProtocols,
 #endif
                         CertificateValidationHandler = _tlsParameters.CertificateValidationHandler,

Source/MQTTnet/Client/Options/MqttClientOptionsBuilderTlsParameters.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Net.Security;
 using System.Security.Authentication;
@@ -31,9 +31,9 @@ namespace MQTTnet.Client.Options
         public IEnumerable<X509Certificate> Certificates { get; set; }
 #endif
 
-#if NETCOREAPP3_1
+#if NETCOREAPP3_1 || NET5_0
 	    public List<SslApplicationProtocol> ApplicationProtocols { get;set; }
-#endif 
+#endif
 
 	    public bool AllowUntrustedCertificates { get; set; }
 

Source/MQTTnet/Client/Options/MqttClientTlsOptions.cs

@@ -22,7 +22,7 @@ namespace MQTTnet.Client.Options
         public List<X509Certificate> Certificates { get; set; }
 #endif
 
-#if NETCOREAPP3_1
+#if NETCOREAPP3_1 || NET5_0
         public List<SslApplicationProtocol> ApplicationProtocols { get; set; }
 #endif
 

Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs

@@ -244,11 +244,6 @@ namespace MQTTnet.Formatter.V3
 
         public MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options)
         {
-            if (options.ReasonCode != MqttClientDisconnectReason.NormalDisconnection || options.ReasonString != null)
-            {
-                throw new MqttProtocolViolationException("Reason codes and reason string for disconnect are only supported for MQTTv5.");
-            }
-
             return new MqttDisconnectPacket();
         }
 

Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs

@@ -75,23 +75,11 @@ namespace MQTTnet.Implementations
                     _socket.Control.IgnorableServerCertificateErrors.Add(ignorableChainValidationResult);
                 }
 
-#if NETCOREAPP3_1 || NET5_0
-                var socketProtectionLevel = SocketProtectionLevel.Tls13;
-                if (_options.TlsOptions.SslProtocol == SslProtocols.Tls12)
-                {
-                    socketProtectionLevel = SocketProtectionLevel.Tls12;
-                }
-                else if (_options.TlsOptions.SslProtocol == SslProtocols.Tls11)
-                {
-                    socketProtectionLevel = SocketProtectionLevel.Tls11;
-                }
-#else
                 var socketProtectionLevel = SocketProtectionLevel.Tls12;
                 if (_options.TlsOptions.SslProtocol == SslProtocols.Tls11)
                 {
                     socketProtectionLevel = SocketProtectionLevel.Tls11;
                 }
-#endif
                 else if (_options.TlsOptions.SslProtocol == SslProtocols.Tls)
                 {
                     socketProtectionLevel = SocketProtectionLevel.Tls10;

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -83,7 +83,7 @@ namespace MQTTnet.Implementations
                     var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
                     try
                     {
-#if NETCOREAPP3_1
+#if NETCOREAPP3_1 || NET5_0
                         var sslOptions = new SslClientAuthenticationOptions
                         {
                             ApplicationProtocols = _tcpOptions.TlsOptions.ApplicationProtocols,
@@ -100,7 +100,7 @@ namespace MQTTnet.Implementations
                     }
                     catch
                     {
-#if NETSTANDARD2_1 || NETCOREAPP3_1
+#if NETSTANDARD2_1 || NETCOREAPP3_1 || NET5_0
                         await sslStream.DisposeAsync().ConfigureAwait(false);
 #else
                         sslStream.Dispose();

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -98,7 +98,7 @@ namespace MQTTnet.Implementations
         {
             _socket?.Dispose();
 
-#if NETSTANDARD1_3 || NETSTANDARD2_0 || NET461 || NET472
+#if !NET452
             _tlsCertificate?.Dispose();
 #endif
         }

Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs

@@ -17,7 +17,7 @@ namespace MQTTnet.Implementations
 
         public static void Sleep(TimeSpan timeout)
         {
-#if NET452 || NETSTANDARD2_0 || NETSTANDARD2_1 || NETCOREAPP3_1
+#if !NETSTANDARD1_3 && !WINDOWS_UWP
             System.Threading.Thread.Sleep(timeout);
 #else
             Task.Delay(timeout).Wait();

Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs

@@ -1,15 +1,30 @@
-using System;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Packets;
 
 namespace MQTTnet
 {
     public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
     {
-        public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage)
+        readonly Func<MqttApplicationMessageReceivedEventArgs, CancellationToken, Task> _acknowledgeHandler;
+        
+        int _isAcknowledged;
+        
+        public MqttApplicationMessageReceivedEventArgs(
+            string clientId, 
+            MqttApplicationMessage applicationMessage,
+            MqttPublishPacket publishPacket,
+            Func<MqttApplicationMessageReceivedEventArgs, CancellationToken, Task> acknowledgeHandler)
         {
             ClientId = clientId;
             ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+            PublishPacket = publishPacket;
+            _acknowledgeHandler = acknowledgeHandler;
         }
 
+        internal MqttPublishPacket PublishPacket { get; }
+        
         /// <summary>
         /// Gets the client identifier.
         /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
@@ -27,7 +42,27 @@ namespace MQTTnet
         /// This value can be used in user code for custom control flow.
         /// </summary>
         public bool IsHandled { get; set; }
+
+        /// <summary>
+        /// Gets ir sets whether the library should send MQTT ACK packets automatically if required.
+        /// </summary>
+        public bool AutoAcknowledge { get; set; } = true;
         
         public object Tag { get; set; }
+        
+        public Task AcknowledgeAsync(CancellationToken cancellationToken) 
+        {
+            if (_acknowledgeHandler == null)
+            {
+                throw new NotSupportedException("Deferred acknowledgement of application message is not yet supported in MQTTnet server.");
+            }
+            
+            if (Interlocked.CompareExchange(ref _isAcknowledged, 1, 0) == 0)
+            {
+                return _acknowledgeHandler(this, cancellationToken);
+            }
+            
+            throw new InvalidOperationException("The application message is already acknowledged.");
+        }
     }
 }

Source/MQTTnet/Server/IMqttServerOptions.cs

@@ -8,7 +8,7 @@ namespace MQTTnet.Server
         /// Gets the client identifier.
         /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
         /// </summary>
-        string ClientId { get; set; }
+        string ClientId { get; }
 
         bool EnablePersistentSessions { get; }
 
@@ -23,16 +23,13 @@ namespace MQTTnet.Server
         IMqttServerUnsubscriptionInterceptor UnsubscriptionInterceptor { get; }
         IMqttServerApplicationMessageInterceptor ApplicationMessageInterceptor { get; }
         IMqttServerClientMessageQueueInterceptor ClientMessageQueueInterceptor { get; }
-
+        IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; }
+        
         MqttServerTcpEndpointOptions DefaultEndpointOptions { get; }
         MqttServerTlsTcpEndpointOptions TlsEndpointOptions { get; }
 
         IMqttServerStorage Storage { get; }
 
         IMqttRetainedMessagesManager RetainedMessagesManager { get; }
-
-        IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; }
-
-        IMqttServerClientDisconnectedHandler ClientDisconnectedInterceptor { get; set; }
     }
 }

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -1,4 +1,4 @@
-using MQTTnet.Adapter;
+using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
@@ -13,7 +13,6 @@ using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Client.Disconnecting;
-using OperationCanceledException = System.OperationCanceledException;
 
 namespace MQTTnet.Server
 {
@@ -21,7 +20,7 @@ namespace MQTTnet.Server
     {
         readonly BlockingCollection<MqttPendingApplicationMessage> _messageQueue = new BlockingCollection<MqttPendingApplicationMessage>();
 
-        readonly object _createConnectionSyncRoot = new object();
+        readonly AsyncLock _createConnectionSyncRoot = new AsyncLock();
         readonly Dictionary<string, MqttClientConnection> _connections = new Dictionary<string, MqttClientConnection>();
         readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
 
@@ -98,7 +97,7 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                var connection = CreateClientConnection(connectPacket, connectionValidatorContext, channelAdapter);
+                var connection = await CreateClientConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false);
                 await _eventDispatcher.SafeNotifyClientConnectedAsync(connectPacket.ClientId).ConfigureAwait(false);
                 await connection.RunAsync().ConfigureAwait(false);
             }
@@ -237,11 +236,13 @@ namespace MQTTnet.Server
                 }
             }
 
+            var endpoint = channelAdapter.Endpoint;
+
             await SafeCleanupChannelAsync(channelAdapter).ConfigureAwait(false);
 
             if (clientId != null)
             {
-                await _eventDispatcher.SafeNotifyClientDisconnectedAsync(clientId, disconnectType).ConfigureAwait(false);
+                await _eventDispatcher.SafeNotifyClientDisconnectedAsync(clientId, disconnectType, endpoint).ConfigureAwait(false);
             }
         }
 
@@ -387,9 +388,12 @@ namespace MQTTnet.Server
             return context;
         }
 
-        MqttClientConnection CreateClientConnection(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
+        async Task<MqttClientConnection> CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
         {
-            lock (_createConnectionSyncRoot)
+            MqttClientConnection existingConnection;
+            MqttClientConnection connection;
+
+            using (await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false))
             {
                 MqttClientSession session;
                 lock (_sessions)
@@ -415,8 +419,6 @@ namespace MQTTnet.Server
                     _sessions[connectPacket.ClientId] = session;
                 }
 
-                MqttClientConnection existingConnection;
-                MqttClientConnection connection;
                 lock (_connections)
                 {
                     _connections.TryGetValue(connectPacket.ClientId, out existingConnection);
@@ -425,10 +427,13 @@ namespace MQTTnet.Server
                     _connections[connectPacket.ClientId] = connection;
                 }
 
-                existingConnection?.StopAsync(MqttClientDisconnectReason.SessionTakenOver).GetAwaiter().GetResult();
+                if (existingConnection != null)
-
+                {
-                return connection;
+                    await existingConnection.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false);
+                }
             }
+
+            return connection;
         }
 
         async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(IMqttServerApplicationMessageInterceptor interceptor, MqttClientConnection clientConnection, MqttApplicationMessage applicationMessage)

Source/MQTTnet/Server/MqttServerClientDisconnectedEventArgs.cs

@@ -1,13 +1,14 @@
-using System;
+using System;
 
 namespace MQTTnet.Server
 {
     public class MqttServerClientDisconnectedEventArgs : EventArgs
     {
-        public MqttServerClientDisconnectedEventArgs(string clientId, MqttClientDisconnectType disconnectType)
+        public MqttServerClientDisconnectedEventArgs(string clientId, MqttClientDisconnectType disconnectType, string endpoint)
         {
             ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
             DisconnectType = disconnectType;
+            Endpoint = endpoint;
         }
 
         /// <summary>
@@ -17,5 +18,7 @@ namespace MQTTnet.Server
         public string ClientId { get; }
 
         public MqttClientDisconnectType DisconnectType { get; }
+
+        public string Endpoint { get; }
     }
 }

Source/MQTTnet/Server/MqttServerEventDispatcher.cs

@@ -1,4 +1,4 @@
-using MQTTnet.Client.Receiving;
+using MQTTnet.Client.Receiving;
 using MQTTnet.Diagnostics;
 using System;
 using System.Threading.Tasks;
@@ -44,7 +44,7 @@ namespace MQTTnet.Server
             }
         }
 
-        public async Task SafeNotifyClientDisconnectedAsync(string clientId, MqttClientDisconnectType disconnectType)
+        public async Task SafeNotifyClientDisconnectedAsync(string clientId, MqttClientDisconnectType disconnectType, string endpoint)
         {
             try
             {
@@ -54,7 +54,7 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                await handler.HandleClientDisconnectedAsync(new MqttServerClientDisconnectedEventArgs(clientId, disconnectType)).ConfigureAwait(false);
+                await handler.HandleClientDisconnectedAsync(new MqttServerClientDisconnectedEventArgs(clientId, disconnectType, endpoint)).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -108,7 +108,7 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)).ConfigureAwait(false);
+                await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage, null, null)).ConfigureAwait(false);
             }
             catch (Exception exception)
             {

Source/MQTTnet/Server/MqttServerOptions.cs

@@ -34,12 +34,10 @@ namespace MQTTnet.Server
 
         public IMqttServerUnsubscriptionInterceptor UnsubscriptionInterceptor { get; set; }
 
+        public IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; }
+
         public IMqttServerStorage Storage { get; set; }
 
         public IMqttRetainedMessagesManager RetainedMessagesManager { get; set; } = new MqttRetainedMessagesManager();
-
-        public IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; }
-
-        public IMqttServerClientDisconnectedHandler ClientDisconnectedInterceptor { get; set; }
     }
 }

Source/MQTTnet/Server/MqttServerOptionsBuilder.cs

@@ -164,19 +164,13 @@ namespace MQTTnet.Server
             _options.ConnectionValidator = new MqttServerConnectionValidatorDelegate(value);
             return this;
         }
-
+        
-        public MqttServerOptionsBuilder WithDisconnectedInterceptor(IMqttServerClientDisconnectedHandler value)
+        public MqttServerOptionsBuilder WithConnectionValidator(Func<MqttConnectionValidatorContext, Task> value)
         {
-            _options.ClientDisconnectedInterceptor = value;
+            _options.ConnectionValidator = new MqttServerConnectionValidatorDelegate(value);
             return this;
         }
-
+        
-        public MqttServerOptionsBuilder WithDisconnectedInterceptor(Action<MqttServerClientDisconnectedEventArgs> value)
-        {
-            _options.ClientDisconnectedInterceptor = new MqttServerClientDisconnectedHandlerDelegate(value);
-            return this;
-        }
-
         public MqttServerOptionsBuilder WithApplicationMessageInterceptor(IMqttServerApplicationMessageInterceptor value)
         {
             _options.ApplicationMessageInterceptor = value;

Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs

@@ -445,6 +445,55 @@ namespace MQTTnet.Tests
             }
         }
 
+        [TestMethod]
+        public async Task Preserve_Message_Order_With_Delayed_Acknowledgement()
+        {
+            // The messages are sent in reverse or to ensure that the delay in the handler
+            // needs longer for the first messages and later messages may be processed earlier (if there
+            // is an issue).
+            const int MessagesCount = 50;
+
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                await testEnvironment.StartServerAsync();
+
+                var client1 = await testEnvironment.ConnectClientAsync();
+                await client1.SubscribeAsync("x", MqttQualityOfServiceLevel.ExactlyOnce);
+
+                var receivedValues = new List<int>();
+
+                Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs)
+                {
+                    var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString());
+                    eventArgs.AutoAcknowledge = false;
+                    Task.Delay(value).ContinueWith(x => eventArgs.AcknowledgeAsync(CancellationToken.None));
+
+                    System.Diagnostics.Debug.WriteLine($"received {value}");
+                    lock (receivedValues)
+                    {
+                        receivedValues.Add(value);
+                    }
+
+                    return Task.CompletedTask;
+                }
+
+                client1.UseApplicationMessageReceivedHandler(Handler1);
+
+                var client2 = await testEnvironment.ConnectClientAsync();
+                for (var i = MessagesCount; i > 0; i--)
+                {
+                    await client2.PublishAsync("x", i.ToString(), MqttQualityOfServiceLevel.ExactlyOnce);
+                }
+
+                await Task.Delay(5000);
+
+                for (var i = MessagesCount; i > 0; i--)
+                {
+                    Assert.AreEqual(i, receivedValues[MessagesCount - i]);
+                }
+            }
+        }
+
         [TestMethod]
         public async Task Send_Reply_For_Any_Received_Message()
         {