Fix handling of retained messages with QoS > 0.

Fix wrong event args types.
Update release notes.
2026-10-16 01:03:01 +08:00 · 2022-06-15 21:56:13 +02:00 · 2022-06-14 21:10:53 +02:00 · 2022-06-10 23:29:52 +02:00 · 2022-06-10 23:27:00 +02:00 · 2022-06-10 23:07:00 +02:00
21 changed files with 262 additions and 183 deletions
@@ -24,7 +24,7 @@ A clear and concise description of what the bug is.
 <!-- Remove the items which don't apply from the following list. -->
 - Client
 - ManagedClient
- RcpClient
+- RpcClient
 - Server

 ### To Reproduce
@@ -1,32 +1,6 @@
-We have joined the .NET Foundation!
-
-Version 4 comes with a new API so a lot of breaking changes should be expected.
-Checkout the upgrade guide (https://github.com/dotnet/MQTTnet/wiki/Upgrading-guide) for an overview of the changes.
-Checkout the new samples (https://github.com/dotnet/MQTTnet/tree/feature/master/Samples) how to use the new API. The wiki only remains for version 3 of this library.
-
-Preview builds of this library are available at: https://www.myget.org/feed/mqttnet/package/nuget/MQTTnet
-
-* [Core] Improved memory management when working with large payloads.
-* [Core] Added support for .NET 6.0.
-* [Core] nuget packages are now created by MSBuild including more information (i.e. commit hash).
-* [Client] Exposed socket linger state in options.
-* [Client] The OS will now choose the best TLS version to use. It is no longer fixed to 1.3 etc. (thanks to @patagonaa, #1271).
-* [Client] Added support for _ServerKeepAlive_ (MQTTv5).
-* [Client] Exposed user properties and reason string in subscribe result.
-* [Client] Exposed user properties and reason string in unsubscribe result.
-* [Client] Migrated application message handler to a regular .NET event (BREAKING CHANGE!).
-* [Client] The will message is longer a regular application message due to not supported properties by the will message (BREAKING CHANGE!).
-* [Client] Timeouts are no longer handled inside the library. Each method (Connect, Publish etc.) supports a cancellation token so that custom timeouts can and must be used (BREAKING CHANGE!).
-* [Client] Exposed certificate revocation mode on options (@andyolivares).
-* [Server] Exposed socket linger state in options.
-* [Server] Added support for returning individual subscription errors (#80 thanks to @jimch)
-* [Server] Improved topic filter comparisons (support for $).
-* [Server] Added more MQTTv5 response information to all interceptors (BREAKING CHANGE!).
-* [Server] Improved session management for MQTT v5 (#1294, thanks to @logicaloud).
-* [Server] All interceptors and events are migrated from interfaces to simple events. All existing APIs are availble but must be migrated to corresponding events (BREAKING CHANGE!).
-* [Server] Removed all interceptor and event interfaces including the delegate implementations etc. (BREAKING CHANGE!).
-* [Server] Renamed a lot of classes and adjusted namespaces (BREAKING CHANGE!).
-* [Server] Introduced a new queueing approach for internal message process (packet bus).
-* [Server] For security reasons the default endpoint (1883) is no longer enabled by default (BREAKING CHANGE!).
-* [Server] Added support for choosing the cipher suite (thanks to @TimSiefert).
-* [Nuget] Added method summaries etc. to nuget packages (thanks to @SpringHgui).
+* [Client] Added the flag _try_private_ (MQTT 3.1.1 Bridge) to client options (#1413).
+* [Client] Fixed MQTTv5 protocol violation in PUBLISH packets (#1423).
+* [Client] Added missing "WithWill..." methods in _MqttClientOptionsBuilder_.
+* [ManagedClient] Fixed wrong event args type for connected and disconnected events (#1432).
+* [Server] Fixed wrong handling of retained messages with QoS > 0 (#1434).
+* [Server] Connections with _try_private_ flag (MQTT 3.1.1 Bridge) are now accepted (#1413).
@@ -3,8 +3,7 @@ name: CI
 on: [push, pull_request]

 env:
-  ASSEMBLY_VERSION: "4.0.0.${{github.run_number}}"
-  NUGET_VERSION:  "4.0.0.${{github.run_number}}"
+  VERSION: "4.0.1.${{github.run_number}}"

 jobs:
  build:
@@ -40,7 +39,7 @@ jobs:
        run: msbuild MQTTnet.sln /t:Restore /p:Configuration="Release" /verbosity:m

      - name: Build solution
-        run: msbuild MQTTnet.sln /t:Build /p:Configuration="Release" /verbosity:m /p:FileVersion=${{ env.ASSEMBLY_VERSION }} /p:AssemblyVersion=${{ env.ASSEMBLY_VERSION }} /p:PackageVersion=${{ env.NUGET_VERSION }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}\certificate.snk
+        run: msbuild MQTTnet.sln /t:Build /p:Configuration="Release" /verbosity:m /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}\certificate.snk

      - name: Collect nuget Packages
        uses: actions/upload-artifact@v2
@@ -8,7 +8,7 @@ using MQTTnet.Protocol;

 namespace MQTTnet.Samples.RpcClient;

-public static class RcpClient_Samples
+public static class RpcClient_Samples
 {
    /*
     * The extension MQTTnet.Extensions.Rpc (available as nuget) allows sending a request and waiting for the matching reply.
@@ -33,7 +33,7 @@ public static class RcpClient_Samples
            
            using (var mqttRpcClient = mqttFactory.CreateMqttRpcClient(mqttClient))
            {
-                // Access to a fully featured application message is not supported for RCP calls!
+                // Access to a fully featured application message is not supported for RPC calls!
                // The method will throw an exception when the response was not received in time.
                await mqttRpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
            }
@@ -81,4 +81,4 @@ public static class RcpClient_Samples
        // ! Do not send response message with the _retain_ flag set to true.
        // ! All required data for a RPC call and the result must be placed into the payload.
     */
-}
+}
@@ -16,7 +16,7 @@
        <PublishRepositoryUrl>true</PublishRepositoryUrl>
        <IncludeSymbols>true</IncludeSymbols>
        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <Copyright>Christian Kratky 2016-2022</Copyright>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
        <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
        <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
        <RepositoryType>git</RepositoryType>
@@ -27,7 +27,7 @@
        <PackageIcon>nuget.png</PackageIcon>
        <EmbedUntrackedSources>true</EmbedUntrackedSources>
        <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
        <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
        <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
        <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
@@ -41,14 +41,7 @@
            <PackagePath>\</PackagePath>
        </None>
    </ItemGroup>
-
-    <ItemGroup>
-        <None Include="..\..\LICENSE">
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
-    </ItemGroup>
-
+    
    <ItemGroup>
        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
    </ItemGroup>
@@ -15,13 +15,13 @@ namespace MQTTnet.Extensions.ManagedClient
        
        event Func<ApplicationMessageSkippedEventArgs, Task> ApplicationMessageSkippedAsync;
        
-        event Func<EventArgs, Task> ConnectedAsync;
+        event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync;
        
        event Func<ConnectingFailedEventArgs, Task> ConnectingFailedAsync;
        
        event Func<EventArgs, Task> ConnectionStateChangedAsync;
        
-        event Func<EventArgs, Task> DisconnectedAsync;
+        event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync;
        
        event Func<ManagedProcessFailedEventArgs, Task> SynchronizingSubscriptionsFailedAsync;
        
@@ -18,7 +18,7 @@
        <PublishRepositoryUrl>true</PublishRepositoryUrl>
        <IncludeSymbols>true</IncludeSymbols>
        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <Copyright>Christian Kratky 2016-2022</Copyright>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
        <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
        <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
        <RepositoryType>git</RepositoryType>
@@ -29,7 +29,7 @@
        <PackageIcon>nuget.png</PackageIcon>
        <EmbedUntrackedSources>true</EmbedUntrackedSources>
        <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
        <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
        <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
        <GenerateDocumentationFile>true</GenerateDocumentationFile>
@@ -55,14 +55,7 @@
            <PackagePath>\</PackagePath>
        </None>
    </ItemGroup>
-
-    <ItemGroup>
-        <None Include="..\..\LICENSE">
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
-    </ItemGroup>
-
+    
    <ItemGroup>
        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
    </ItemGroup>
@@ -93,7 +93,7 @@ namespace MQTTnet.Extensions.ManagedClient
            remove => InternalClient.ApplicationMessageReceivedAsync -= value;
        }

-        public event Func<EventArgs, Task> ConnectedAsync
+        public event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync
        {
            add => InternalClient.ConnectedAsync += value;
            remove => InternalClient.ConnectedAsync -= value;
@@ -111,7 +111,7 @@ namespace MQTTnet.Extensions.ManagedClient
            remove => _connectionStateChangedEvent.RemoveHandler(value);
        }

-        public event Func<EventArgs, Task> DisconnectedAsync
+        public event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync
        {
            add => InternalClient.DisconnectedAsync += value;
            remove => InternalClient.DisconnectedAsync -= value;
@@ -18,7 +18,7 @@
        <PublishRepositoryUrl>true</PublishRepositoryUrl>
        <IncludeSymbols>true</IncludeSymbols>
        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <Copyright>Christian Kratky 2016-2022</Copyright>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
        <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
        <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
        <RepositoryType>git</RepositoryType>
@@ -29,7 +29,7 @@
        <PackageIcon>nuget.png</PackageIcon>
        <EmbedUntrackedSources>true</EmbedUntrackedSources>
        <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
        <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
        <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
        <GenerateDocumentationFile>true</GenerateDocumentationFile>
@@ -55,14 +55,7 @@
            <PackagePath>\</PackagePath>
        </None>
    </ItemGroup>
-
-    <ItemGroup>
-        <None Include="..\..\LICENSE">
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
-    </ItemGroup>
-
+    
    <ItemGroup>
        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
    </ItemGroup>
@@ -18,7 +18,7 @@
        <PublishRepositoryUrl>true</PublishRepositoryUrl>
        <IncludeSymbols>true</IncludeSymbols>
        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <Copyright>Christian Kratky 2016-2022</Copyright>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
        <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
        <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
        <RepositoryType>git</RepositoryType>
@@ -29,7 +29,7 @@
        <PackageIcon>nuget.png</PackageIcon>
        <EmbedUntrackedSources>true</EmbedUntrackedSources>
        <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
        <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
        <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
        <GenerateDocumentationFile>true</GenerateDocumentationFile>
@@ -55,14 +55,7 @@
            <PackagePath>\</PackagePath>
        </None>
    </ItemGroup>
-
-    <ItemGroup>
-        <None Include="..\..\LICENSE">
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
-    </ItemGroup>
-
+    
    <ItemGroup>
        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
    </ItemGroup>
@@ -16,67 +16,52 @@ namespace MQTTnet.Tests.Server
    public sealed class Retained_Messages_Tests : BaseTestClass
    {
        [TestMethod]
-        public async Task Server_Reports_Retained_Messages_Supported_V3()
+        public async Task Clear_Retained_Message_With_Empty_Payload()
        {
            using (var testEnvironment = CreateTestEnvironment())
            {
                await testEnvironment.StartServer();

-                var client = testEnvironment.CreateClient();
-                
-                var connectResult = await client.ConnectAsync(testEnvironment.Factory.CreateClientOptionsBuilder()
-                    .WithProtocolVersion(MqttProtocolVersion.V311)
-                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
-
-                Assert.IsTrue(connectResult.RetainAvailable);
-            }
-        }
-        
-        [TestMethod]
-        public async Task Server_Reports_Retained_Messages_Supported_V5()
-        {
-            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
-            {
-                await testEnvironment.StartServer();
-
-                var client = testEnvironment.CreateClient();
-                var connectResult = await client.ConnectAsync(testEnvironment.Factory.CreateClientOptionsBuilder()
-                    .WithProtocolVersion(MqttProtocolVersion.V500)
-                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
-
-                Assert.IsTrue(connectResult.RetainAvailable);
-            }
-        }
-        
-        [TestMethod]
-        public async Task Retained_Messages_Flow()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                var retainedMessage = new MqttApplicationMessageBuilder().WithTopic("r").WithPayload("r").WithRetainFlag().Build();
-
-                await testEnvironment.StartServer();
                var c1 = await testEnvironment.ConnectClient();
-    
+
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build());
+
+                await c1.DisconnectAsync();
+
                var c2 = await testEnvironment.ConnectClient();
                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                
-                await c1.PublishAsync(retainedMessage);
+
+                await Task.Delay(200);
+                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                await Task.Delay(500);
+
+                messageHandler.AssertReceivedCountEquals(0);
+            }
+        }
+
+        [TestMethod]
+        public async Task Clear_Retained_Message_With_Null_Payload()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var c1 = await testEnvironment.ConnectClient();
+
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload((byte[])null).WithRetainFlag().Build());
+
                await c1.DisconnectAsync();
-                await LongTestDelay();

-                for (var i = 0; i < 5; i++)
-                {
-                    await c2.UnsubscribeAsync("r");
-                    await Task.Delay(100);
-                    messageHandler.AssertReceivedCountEquals(i);
-                    
-                    await c2.SubscribeAsync("r");
-                    await Task.Delay(100);
-                    messageHandler.AssertReceivedCountEquals(i + 1);
-                }
+                var c2 = await testEnvironment.ConnectClient();
+                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);

-                await c2.DisconnectAsync();
+                await Task.Delay(200);
+                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                await Task.Delay(500);
+
+                messageHandler.AssertReceivedCountEquals(0);
            }
        }

@@ -114,7 +99,7 @@ namespace MQTTnet.Tests.Server

                var c2 = await testEnvironment.ConnectClient();
                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                
+
                await c2.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("retained").Build());

                await Task.Delay(500);
@@ -125,52 +110,102 @@ namespace MQTTnet.Tests.Server
        }

        [TestMethod]
-        public async Task Clear_Retained_Message_With_Empty_Payload()
+        public async Task Receive_Retained_Messages_From_Higher_Qos_Level()
        {
            using (var testEnvironment = CreateTestEnvironment())
            {
                await testEnvironment.StartServer();

+                // Upload retained message.
                var c1 = await testEnvironment.ConnectClient();
-
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build());
-
+                await c1.PublishAsync(
+                    new MqttApplicationMessageBuilder().WithTopic("retained")
+                        .WithPayload(new byte[1])
+                        .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
+                        .WithRetainFlag()
+                        .Build());
+                
                await c1.DisconnectAsync();

+                // Subscribe using a new client.
                var c2 = await testEnvironment.ConnectClient();
                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                
+
                await Task.Delay(200);
-                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                // Using QoS 2 will lead to 1 instead because the publish was made with QoS level 1 (see 3.8.4 SUBSCRIBE Actions)!
+                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce });
                await Task.Delay(500);

-                messageHandler.AssertReceivedCountEquals(0);
+                messageHandler.AssertReceivedCountEquals(1);
            }
        }

        [TestMethod]
-        public async Task Clear_Retained_Message_With_Null_Payload()
+        public async Task Retained_Messages_Flow()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var retainedMessage = new MqttApplicationMessageBuilder().WithTopic("r").WithPayload("r").WithRetainFlag().Build();
+
+                await testEnvironment.StartServer();
+                var c1 = await testEnvironment.ConnectClient();
+
+                var c2 = await testEnvironment.ConnectClient();
+                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
+
+                await c1.PublishAsync(retainedMessage);
+                await c1.DisconnectAsync();
+                await LongTestDelay();
+
+                for (var i = 0; i < 5; i++)
+                {
+                    await c2.UnsubscribeAsync("r");
+                    await Task.Delay(100);
+                    messageHandler.AssertReceivedCountEquals(i);
+
+                    await c2.SubscribeAsync("r");
+                    await Task.Delay(100);
+                    messageHandler.AssertReceivedCountEquals(i + 1);
+                }
+
+                await c2.DisconnectAsync();
+            }
+        }
+
+        [TestMethod]
+        public async Task Server_Reports_Retained_Messages_Supported_V3()
        {
            using (var testEnvironment = CreateTestEnvironment())
            {
                await testEnvironment.StartServer();

-                var c1 = await testEnvironment.ConnectClient();
+                var client = testEnvironment.CreateClient();

-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload((byte[])null).WithRetainFlag().Build());
+                var connectResult = await client.ConnectAsync(
+                    testEnvironment.Factory.CreateClientOptionsBuilder()
+                        .WithProtocolVersion(MqttProtocolVersion.V311)
+                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                        .Build());

-                await c1.DisconnectAsync();
+                Assert.IsTrue(connectResult.RetainAvailable);
+            }
+        }

-                var c2 = await testEnvironment.ConnectClient();
-                var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
-                
-                await Task.Delay(200);
-                await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
-                await Task.Delay(500);
+        [TestMethod]
+        public async Task Server_Reports_Retained_Messages_Supported_V5()
+        {
+            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
+            {
+                await testEnvironment.StartServer();

-                messageHandler.AssertReceivedCountEquals(0);
+                var client = testEnvironment.CreateClient();
+                var connectResult = await client.ConnectAsync(
+                    testEnvironment.Factory.CreateClientOptionsBuilder()
+                        .WithProtocolVersion(MqttProtocolVersion.V500)
+                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                        .Build());
+
+                Assert.IsTrue(connectResult.RetainAvailable);
            }
        }
    }
@@ -98,6 +98,16 @@ namespace MQTTnet.Client
        /// </summary>
        public ushort TopicAliasMaximum { get; set; }

+        /// <summary>
+        ///     If set to true, the bridge will attempt to indicate to the remote broker that it is a bridge not an ordinary
+        ///     client.
+        ///     If successful, this means that loop detection will be more effective and that retained messages will be propagated
+        ///     correctly.
+        ///     Not all brokers support this feature so it may be necessary to set it to false if your bridge does not connect
+        ///     properly.
+        /// </summary>
+        public bool TryPrivate { get; set; } = true;
+
        /// <summary>
        ///     Gets or sets the user properties.
        ///     In MQTT 5, user properties are basic UTF-8 string key-value pairs that you can append to almost every type of MQTT
@@ -96,16 +96,6 @@ namespace MQTTnet.Client
            return this;
        }

-        /// <summary>
-        ///     Sets the timeout which will be applied at socket level and internal operations.
-        ///     The default value is the same as for sockets in .NET in general.
-        /// </summary>
-        public MqttClientOptionsBuilder WithTimeout(TimeSpan value)
-        {
-            _options.Timeout = value;
-            return this;
-        }
-
        public MqttClientOptionsBuilder WithConnectionUri(Uri uri)
        {
            if (uri == null)
@@ -289,6 +279,16 @@ namespace MQTTnet.Client
            return this;
        }

+        /// <summary>
+        ///     Sets the timeout which will be applied at socket level and internal operations.
+        ///     The default value is the same as for sockets in .NET in general.
+        /// </summary>
+        public MqttClientOptionsBuilder WithTimeout(TimeSpan value)
+        {
+            _options.Timeout = value;
+            return this;
+        }
+
        public MqttClientOptionsBuilder WithTls(MqttClientOptionsBuilderTlsParameters parameters)
        {
            _tlsParameters = parameters;
@@ -322,6 +322,20 @@ namespace MQTTnet.Client
            return this;
        }

+        /// <summary>
+        ///     If set to true, the bridge will attempt to indicate to the remote broker that it is a bridge not an ordinary
+        ///     client.
+        ///     If successful, this means that loop detection will be more effective and that retained messages will be propagated
+        ///     correctly.
+        ///     Not all brokers support this feature so it may be necessary to set it to false if your bridge does not connect
+        ///     properly.
+        /// </summary>
+        public MqttClientOptionsBuilder WithTryPrivate(bool tryPrivate = true)
+        {
+            _options.TryPrivate = true;
+            return this;
+        }
+
        public MqttClientOptionsBuilder WithUserProperty(string name, string value)
        {
            if (name is null)
@@ -379,6 +393,17 @@ namespace MQTTnet.Client
            _options.WillPayload = willPayload;
            return this;
        }
+        
+        public MqttClientOptionsBuilder WithWillPayload(string willPayload)
+        {
+            if (string.IsNullOrEmpty(willPayload))
+            {
+                return WithWillPayload((byte[])null);
+            }
+            
+            _options.WillPayload = Encoding.UTF8.GetBytes(willPayload);
+            return this;
+        }

        public MqttClientOptionsBuilder WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel willQualityOfServiceLevel)
        {
@@ -391,5 +416,46 @@ namespace MQTTnet.Client
            _options.WillTopic = willTopic;
            return this;
        }
+
+        public MqttClientOptionsBuilder WithWillRetain(bool willRetain = true)
+        {
+            _options.WillRetain = willRetain;
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithWillContentType(string willContentType)
+        {
+            _options.WillContentType = willContentType;
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithWillCorrelationData(byte[] willCorrelationData)
+        {
+            _options.WillCorrelationData = willCorrelationData;
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithWillResponseTopic(string willResponseTopic)
+        {
+            _options.WillResponseTopic = willResponseTopic;
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithWillPayloadFormatIndicator(MqttPayloadFormatIndicator willPayloadFormatIndicator)
+        {
+            _options.WillPayloadFormatIndicator = willPayloadFormatIndicator;
+            return this;
+        }
+        
+        public MqttClientOptionsBuilder WithWillUserProperty(string name, string value)
+        {
+            if (_options.WillUserProperties == null)
+            {
+                _options.WillUserProperties = new List<MqttUserProperty>();
+            }
+            
+            _options.WillUserProperties.Add(new MqttUserProperty(name, value));
+            return this;
+        }
    }
 }
@@ -33,7 +33,8 @@ namespace MQTTnet.Formatter
                RequestResponseInformation = clientOptions.RequestResponseInformation,
                SessionExpiryInterval = clientOptions.SessionExpiryInterval,
                TopicAliasMaximum = clientOptions.TopicAliasMaximum,
-                UserProperties = clientOptions.UserProperties
+                UserProperties = clientOptions.UserProperties,
+                TryPrivate = clientOptions.TryPrivate
            };

            if (!string.IsNullOrEmpty(clientOptions.WillTopic))
@@ -95,6 +95,10 @@ namespace MQTTnet.Formatter
            var protocolName = _bufferReader.ReadString();
            var protocolLevel = _bufferReader.ReadByte();

+            // Remove the mosquitto try_private flag (MQTT 3.1.1 Bridge).
+            // This flag is accepted but not yet used.
+            protocolLevel &= 0x7F;
+            
            if (protocolName == "MQTT")
            {
                if (protocolLevel == 5)
@@ -175,12 +175,18 @@ namespace MQTTnet.Formatter.V3
                throw new MqttProtocolViolationException("MQTT protocol name do not match MQTT v3.");
            }

+            var tryPrivate = (protocolVersion & 0x80) > 0;
+            protocolVersion &= 0x7F;
+            
            if (protocolVersion != 3 && protocolVersion != 4)
            {
                throw new MqttProtocolViolationException("MQTT protocol version do not match MQTT v3.");
            }

-            var packet = new MqttConnectPacket();
+            var packet = new MqttConnectPacket
+            {
+                TryPrivate = tryPrivate
+            };

            var connectFlags = _bufferReader.ReadByte();
            if ((connectFlags & 0x1) > 0)
@@ -409,7 +415,14 @@ namespace MQTTnet.Formatter.V3
            ValidateConnectPacket(packet);

            bufferWriter.WriteString("MQIsdp");
-            bufferWriter.WriteByte(3); // Protocol Level 3
+
+            var protocolVersion = 3;
+            if (packet.TryPrivate)
+            {
+                protocolVersion |= 0x80;
+            }
+            
+            bufferWriter.WriteByte((byte)protocolVersion);

            byte connectFlags = 0x0;
            if (packet.CleanSession)
@@ -84,6 +84,11 @@ namespace MQTTnet.Formatter.V5
        {
            // If absent, the Application Message does not expire.
            // This library uses 0 to indicate no expiration.
+            if (value == 0)
+            {
+                return;
+            }
+            
            WriteAsFourByteInteger(MqttPropertyId.MessageExpiryInterval, value);
        }

@@ -29,7 +29,7 @@
        <PublishRepositoryUrl>true</PublishRepositoryUrl>
        <IncludeSymbols>true</IncludeSymbols>
        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <Copyright>Christian Kratky 2016-2022</Copyright>
+        <Copyright>Copyright (c) .NET Foundation and Contributors</Copyright>
        <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
        <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
        <RepositoryType>git</RepositoryType>
@@ -40,7 +40,7 @@
        <PackageIcon>nuget.png</PackageIcon>
        <EmbedUntrackedSources>true</EmbedUntrackedSources>
        <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
        <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
        <NoWarn>1591</NoWarn>
    </PropertyGroup>
@@ -75,13 +75,6 @@
        </None>
    </ItemGroup>

-    <ItemGroup>
-        <None Include="..\..\LICENSE">
-            <Pack>True</Pack>
-            <PackagePath>\</PackagePath>
-        </None>
-    </ItemGroup>
-
    <ItemGroup>
        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
    </ItemGroup>
@@ -64,6 +64,8 @@ namespace MQTTnet.Packets

        public List<MqttUserProperty> WillUserProperties { get; set; }

+        public bool TryPrivate { get; set; }
+        
        public override string ToString()
        {
            var passwordText = string.Empty;
@@ -392,7 +392,7 @@ namespace MQTTnet.Server
                foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages)
                {
                    var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
-                    Session.EnqueueControlPacket(new MqttPacketBusItem(publishPacket));
+                    Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
                }
            }
        }
@@ -19,6 +19,7 @@ namespace MQTTnet.Server
    {
        readonly MqttClientSessionsManager _clientSessionsManager;
        readonly MqttPacketBus _packetBus = new MqttPacketBus();
+        readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();

        readonly MqttServerOptions _serverOptions;

@@ -100,6 +101,11 @@ namespace MQTTnet.Server
            SubscriptionsManager.Dispose();
        }

+        public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)
+        {
+            _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
+        }
+
        public void EnqueueDataPacket(MqttPacketBusItem packetBusItem)
        {
            if (_packetBus.ItemsCount(MqttPacketBusPartition.Data) >= _serverOptions.MaxPendingMessagesPerClient)
@@ -118,28 +124,27 @@ namespace MQTTnet.Server
            }

            var publishPacket = (MqttPublishPacket)packetBusItem.Packet;
+
            if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
            {
+                publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
+
                _unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket;
            }

            _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
        }

-        public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)
-        {
-            _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
-        }
-
        public void EnqueueHealthPacket(MqttPacketBusItem packetBusItem)
        {
            _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health);
        }
-        
+
        public void Recover()
        {
            // TODO: Keep the bus and only insert pending items again.
            // TODO: Check if packet identifier must be restarted or not.
+            // TODO: Recover package identifier.
            _packetBus.Clear();

            foreach (var publishPacket in _unacknowledgedPublishPackets.Values.ToList())
Author	SHA1	Message	Date
Christian	a3c7f9a22c	Fix handling of retained messages with QoS > 0.	2022-06-15 21:56:13 +02:00
Christian	41e0a25fc8	Fix wrong event args types.	2022-06-14 21:10:53 +02:00
Christian	772cd3cd12	Update release notes.	2022-06-10 23:29:52 +02:00
Christian	42cf9f73c2	Merge branch 'master' into 1413-protocol-version-132-mqtt-311-bridge-support	2022-06-10 23:27:00 +02:00
Christian	ec2e08668f	Add support for MQTT 3.1.1 Bridge flag.	2022-06-10 23:07:00 +02:00
Christian	e9e8583d1a	Merge branch 'master' of https://github.com/dotnet/MQTTnet	2022-06-10 17:10:14 +02:00
Christian	f68b109471	Switch to license expression. (#1408 )	2022-06-10 17:09:17 +02:00
Christian	dd1e155ff4	Add missing "WithWill..." methods.	2022-06-09 21:52:05 +02:00
Christian	cf6369d63f	Update version.	2022-06-09 21:27:14 +02:00
Christian	3049fc6608	Update ReleaseNotes.md	2022-06-09 21:27:08 +02:00
Christian	b75076a1c5	Skip adding default message expiry interval.	2022-06-09 21:23:34 +02:00
Christian	ae8e2ec7c9	Merge branch 'master' of https://github.com/dotnet/MQTTnet	2022-06-09 20:13:19 +02:00
Christian	e7e6a080e0	Update copyright information.	2022-06-09 20:13:17 +02:00
SeppPenner	dc9afa6c3a	Samples, Github template: Fixed small typos.	2022-06-08 09:19:14 +02:00
HansM	02b17f277d	Merge pull request #1421 from soyoo/patch-1 Update RcpClient_Samples.cs	2022-06-08 08:58:22 +02:00
soyoo	b9d9ea2d4d	Update RcpClient_Samples.cs typo	2022-06-07 21:54:08 +08:00