Backport of: 2132 Fix support for enhanced authentication (#2171)

* Increase version

* Update release notes

* Adjust cancellation token handling

* Fix CI build

.github/workflows/ReleaseNotes.md

@@ -1 +1,3 @@
-* [Client] Restored _Server_ and _Port_ behavior of client options (#2005).
+* Client: Fixed wrong timeout for keep alive check (thanks to @Erw1nT, #2129)
+* Client: Fixed extended authentication (thanks to @chkr1011, #2132).
+* Server: Fixed extended authentication (thanks to @chkr1011, #2132).

.github/workflows/ci.yml

@@ -3,7 +3,7 @@ name: CI
 on: [push, pull_request]
 
 env:
-  VERSION: "4.3.6.${{github.run_number}}"
+  VERSION: "4.3.8.${{github.run_number}}"
 
 jobs:
   build:
@@ -12,12 +12,12 @@ jobs:
 
     steps:
       - name: Setup Windows SDK
-        uses: GuillaumeFalourd/setup-windows10-sdk-action@v1
+        uses: GuillaumeFalourd/setup-windows10-sdk-action@v2.4
         with:
           sdk-version: 18362
 
       - name: Setup .NET SDK
-        uses: actions/setup-dotnet@v1.9.0
+        uses: actions/setup-dotnet@v4
         with:
           dotnet-version: |
             3.1.x
@@ -28,7 +28,7 @@ jobs:
         uses: microsoft/setup-msbuild@v1.1
 
       - name: Checkout Code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v4
 
       - name: Setup Signing Certificate
         run: |
@@ -42,14 +42,6 @@ jobs:
       - name: Build solution
         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
-        with:
-          name: nuget Packages
-          path: |
-            **\*.nupkg
-            **\*.snupkg
-
       - name: Setup VSTest
         uses: darenm/Setup-VSTest@v1
 
@@ -59,6 +51,32 @@ jobs:
       - name: ASP.NET Tests
         run: vstest.console.exe Source\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\MQTTnet.AspNetCore.Tests.dll
 
+      - name: Install sign CLI tool
+        run: dotnet tool install --tool-path . sign --version 0.9.0-beta.23127.3
+
+      - name: Sign nugets
+        shell: pwsh
+        run: >
+          ./sign code azure-key-vault `
+          "**/*.nupkg" `
+          --base-directory "${{ github.workspace }}/Source" `
+          --publisher-name "MQTTnet" `
+          --description "MQTTnet" `
+          --description-url "https://github.com/dotnet/MQTTnet" `
+          --azure-key-vault-tenant-id "${{ secrets.AZURE_TENANT_ID }}" `
+          --azure-key-vault-client-id "${{ secrets.AZURE_CLIENT_ID }}" `
+          --azure-key-vault-client-secret "${{ secrets.AZURE_CLIENT_SECRET }}" `
+          --azure-key-vault-certificate "${{ secrets.KEY_VAULT_CERTIFICATE_ID }}" `
+          --azure-key-vault-url "${{ secrets.KEY_VAULT_URL }}"
+
+      - name: Upload signed nuget packages
+        uses: actions/upload-artifact@v4
+        with:
+          name: signed-nugets
+          path: |
+            ${{ github.workspace }}/Source/**/*.nupkg
+            ${{ github.workspace }}/Source/**/*.snupkg
+
       - name: Publish MyGet nugets
         if: ${{ github.event_name == 'push' }}
-        run: dotnet nuget push **/*.nupkg -k ${{ secrets.MYGET_API_KEY }} -s https://www.myget.org/F/mqttnet/api/v3/index.json --skip-duplicate
+        run: dotnet nuget push ${{ github.workspace }}/Source/**/*.nupkg -k ${{ secrets.MYGET_API_KEY }} -s https://www.myget.org/F/mqttnet/api/v3/index.json --skip-duplicate

MQTTnet.sln

@@ -10,6 +10,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
 		CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md
 		LICENSE = LICENSE
 		README.md = README.md
+		.github\workflows\ReleaseNotes.md = .github\workflows\ReleaseNotes.md
+		.github\workflows\ci.yml = .github\workflows\ci.yml
 	EndProjectSection
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}"

Samples/Client/Client_Connection_Samples.cs

@@ -8,10 +8,13 @@
 
 using System.Security.Authentication;
 using System.Security.Cryptography.X509Certificates;
+using System.Text;
 using MQTTnet.Client;
 using MQTTnet.Extensions.WebSocket4Net;
 using MQTTnet.Formatter;
+using MQTTnet.Protocol;
 using MQTTnet.Samples.Helpers;
+using MQTTnet.Server;
 
 namespace MQTTnet.Samples.Client;
 
@@ -100,6 +103,105 @@ public static class Client_Connection_Samples
         }
     }
 
+    public static async Task Connect_Using_Extended_Authentication()
+    {
+        /*
+         * This sample uses enhanced (aka extended) authentication (Kerberos) when creating the connection.
+         */
+
+        /*
+         * Server part...
+         */
+
+        var mqttFactory = new MqttFactory();
+
+        var serverOptions = mqttFactory.CreateServerOptionsBuilder().WithDefaultEndpoint().Build();
+        var server = mqttFactory.CreateMqttServer(serverOptions);
+
+        server.ValidatingConnectionAsync += async args =>
+        {
+            if (args.AuthenticationMethod == "GS2-KRB5")
+            {
+                var result = await args.SendExtendedAuthenticationExchangeDataAsync(new MqttExtendedAuthenticationExchangeOptions());
+
+                Console.WriteLine($"Received AUTH data from client: {Encoding.UTF8.GetString(result.AuthenticationData)}");
+
+                var authOptions = mqttFactory.CreateMqttExtendedAuthenticationExchangeOptionsBuilder().WithAuthenticationData("reply context token").Build();
+
+                result = await args.SendExtendedAuthenticationExchangeDataAsync(authOptions);
+
+                Console.WriteLine($"Received AUTH data from client: {Encoding.UTF8.GetString(result.AuthenticationData)}");
+
+                args.ResponseAuthenticationData = "outcome of authentication"u8.ToArray();
+
+                // Authentication DONE!
+                args.ReasonCode = MqttConnectReasonCode.Success; // Also the default!
+            }
+            else
+            {
+                args.ReasonCode = MqttConnectReasonCode.BadAuthenticationMethod;
+            }
+        };
+
+        await server.StartAsync();
+
+        /*
+         * Client part...
+         */
+
+        var mqttClientFactory = new MqttFactory();
+
+        // Use Kerberos sample from the MQTT RFC.
+        var kerberosAuthenticationHandler = new SampleClientKerberosAuthenticationHandler();
+
+        var clientOptions = mqttClientFactory.CreateClientOptionsBuilder()
+            .WithTcpServer("localhost")
+            .WithProtocolVersion(MqttProtocolVersion.V500)
+            .WithExtendedAuthenticationExchangeHandler(kerberosAuthenticationHandler)
+            .WithAuthentication("GS2-KRB5", Array.Empty<byte>())
+            .Build();
+
+        var client = mqttClientFactory.CreateMqttClient();
+
+        var result = await client.ConnectAsync(clientOptions);
+
+        Console.WriteLine($"Client connect result: {result.ResultCode}");
+    }
+
+    class SampleClientKerberosAuthenticationHandler : IMqttExtendedAuthenticationExchangeHandler
+    {
+        private readonly Queue<byte[]> clientTokens = new Queue<byte[]>();
+
+        public SampleClientKerberosAuthenticationHandler()
+        {
+            // The Kerberos sample from the MQTT RFC sends two client tokens after the initial empty token.
+            clientTokens.Enqueue(Encoding.UTF8.GetBytes("initial context token"));
+            clientTokens.Enqueue(Array.Empty<byte>());
+        }
+
+        public async Task HandleRequestAsync(MqttExtendedAuthenticationExchangeContext context)
+        {
+            if (context.AuthenticationMethod != "GS2-KRB5")
+            {
+                throw new InvalidOperationException("Wrong authentication method");
+            }
+
+            Console.WriteLine($"Received AUTH data from server: {Encoding.UTF8.GetString(context.AuthenticationData)}");
+
+            if (!clientTokens.TryDequeue(out var clientToken))
+            {
+                throw new InvalidOperationException("No more client tokens available");
+            }
+
+            var sendOptions = new MqttExtendedAuthenticationExchangeData
+            {
+                AuthenticationData = clientToken
+            };
+
+            await context.Client.SendExtendedAuthenticationExchangeDataAsync(sendOptions);
+        }
+    }
+
     public static async Task Connect_Client_Using_MQTTv5()
     {
         /*

Samples/Client/Client_Subscribe_Samples.cs

@@ -8,6 +8,7 @@
 // ReSharper disable UnusedMember.Local
 
 using MQTTnet.Client;
+using MQTTnet.Extensions.TopicTemplate;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Samples.Helpers;
@@ -16,6 +17,8 @@ namespace MQTTnet.Samples.Client;
 
 public static class Client_Subscribe_Samples
 {
+    static MqttTopicTemplate sampleTemplate = new MqttTopicTemplate("mqttnet/samples/topic/{id}");
+    
     public static async Task Handle_Received_Application_Message()
     {
         /*
@@ -42,11 +45,7 @@ public static class Client_Subscribe_Samples
             await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
             var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
-                .WithTopicFilter(
-                    f =>
-                    {
-                        f.WithTopic("mqttnet/samples/topic/2");
-                    })
+                .WithTopicTemplate(sampleTemplate.WithParameter("id", "2"))
                 .Build();
 
             await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -88,11 +87,8 @@ public static class Client_Subscribe_Samples
             await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
             var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
-                .WithTopicFilter(
-                    f =>
-                    {
-                        f.WithTopic("mqttnet/samples/topic/1");
-                    })
+                .WithTopicTemplate(
+                    sampleTemplate.WithParameter("id", "1"))
                 .Build();
 
             var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -121,21 +117,12 @@ public static class Client_Subscribe_Samples
             // Create the subscribe options including several topics with different options.
             // It is also possible to all of these topics using a dedicated call of _SubscribeAsync_ per topic.
             var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
-                .WithTopicFilter(
-                    f =>
-                    {
-                        f.WithTopic("mqttnet/samples/topic/1");
-                    })
-                .WithTopicFilter(
-                    f =>
-                    {
-                        f.WithTopic("mqttnet/samples/topic/2").WithNoLocal();
-                    })
-                .WithTopicFilter(
-                    f =>
-                    {
-                        f.WithTopic("mqttnet/samples/topic/3").WithRetainHandling(MqttRetainHandling.SendAtSubscribe);
-                    })
+                .WithTopicTemplate(
+                    sampleTemplate.WithParameter("id", "1"))
+                .WithTopicTemplate(
+                    sampleTemplate.WithParameter("id", "2"), noLocal: true)
+                .WithTopicTemplate(
+                    sampleTemplate.WithParameter("id", "3"), retainHandling: MqttRetainHandling.SendAtSubscribe)
                 .Build();
 
             var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -162,11 +149,7 @@ public static class Client_Subscribe_Samples
             await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
             var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
-                .WithTopicFilter(
-                    f =>
-                    {
-                        f.WithTopic("mqttnet/samples/topic/1");
-                    })
+                .WithTopicTemplate(sampleTemplate.WithParameter("id", "1"))
                 .Build();
 
             var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);

Samples/MQTTnet.Samples.csproj

@@ -17,6 +17,7 @@
         <ProjectReference Include="..\Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj" />
         <ProjectReference Include="..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj" />
         <ProjectReference Include="..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj" />
+        <ProjectReference Include="..\Source\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj" />
         <ProjectReference Include="..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj" />
         <ProjectReference Include="..\Source\MQTTnet\MQTTnet.csproj" />
     </ItemGroup>

Samples/ManagedClient/Managed_Client_Subscribe_Samples.cs

@@ -22,7 +22,7 @@ public sealed class Managed_Client_Subscribe_Samples
          * - Reconnecting when connection is lost.
          * - Storing pending messages in an internal queue so that an enqueue is possible while the client remains not connected.
          */
-        
+
         var mqttFactory = new MqttFactory();
         var subscribed = false;
 
@@ -43,21 +43,21 @@ public sealed class Managed_Client_Subscribe_Samples
             await managedMqttClient.EnqueueAsync("Topic", "Payload");
 
             Console.WriteLine("The managed MQTT client is connected.");
-            
+
             // Wait until the queue is fully processed.
             SpinWait.SpinUntil(() => managedMqttClient.PendingApplicationMessagesCount == 0, 10000);
-            
+
             Console.WriteLine($"Pending messages = {managedMqttClient.PendingApplicationMessagesCount}");
 
-            managedMqttClient.SubscriptionsChangedAsync += args => SubscriptionsResultAsync(args, subscribed);
+            managedMqttClient.SubscriptionsChangedAsync += args => SubscriptionsResultAsync(args, ref subscribed);
             await managedMqttClient.SubscribeAsync("Topic").ConfigureAwait(false);
 
             SpinWait.SpinUntil(() => subscribed, 1000);
-            Console.WriteLine("Subscription properly done");
+            Console.WriteLine($"Subscription properly done {subscribed}");
         }
     }
 
-    private static Task SubscriptionsResultAsync(SubscriptionsChangedEventArgs arg, bool subscribed)
+    private static Task SubscriptionsResultAsync(SubscriptionsChangedEventArgs arg, ref bool subscribed)
     {
         foreach (var mqttClientSubscribeResult in arg.SubscribeResult)
         {

Source/MQTTnet.AspnetCore/MqttHostedServer.cs

@@ -16,26 +16,51 @@ namespace MQTTnet.AspNetCore
     public sealed class MqttHostedServer : MqttServer, IHostedService
     {
         readonly MqttFactory _mqttFactory;
-
-        public MqttHostedServer(MqttFactory mqttFactory, MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(
+#if NETCOREAPP3_1_OR_GREATER
+        readonly IHostApplicationLifetime _hostApplicationLifetime;
+        public MqttHostedServer(IHostApplicationLifetime hostApplicationLifetime, MqttFactory mqttFactory,
+            MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(
+            options,
+            adapters,
+            logger)
+        {
+            _mqttFactory = mqttFactory ?? throw new ArgumentNullException(nameof(mqttFactory));
+            _hostApplicationLifetime = hostApplicationLifetime;
+        }
+#else
+        public MqttHostedServer(MqttFactory mqttFactory,
+            MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(
             options,
             adapters,
             logger)
         {
             _mqttFactory = mqttFactory ?? throw new ArgumentNullException(nameof(mqttFactory));
         }
+#endif
+
 
         public async Task StartAsync(CancellationToken cancellationToken)
         {
             // The yield makes sure that the hosted service is considered up and running.
             await Task.Yield();
-
+#if NETCOREAPP3_1_OR_GREATER
+            _hostApplicationLifetime.ApplicationStarted.Register(OnStarted);
+#else
             _ = StartAsync();
+#endif
+
         }
 
         public Task StopAsync(CancellationToken cancellationToken)
         {
             return StopAsync(_mqttFactory.CreateMqttServerStopOptionsBuilder().Build());
         }
+
+#if NETCOREAPP3_1_OR_GREATER
+        private void OnStarted()
+        {
+            _ = StartAsync();
+        }
+#endif
     }
 }

Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs

@@ -47,8 +47,8 @@ namespace MQTTnet.Extensions.ManagedClient
         
         Task StopAsync(bool cleanDisconnect = true);
         
-        Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters);
+        Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters);
         
-        Task UnsubscribeAsync(ICollection<string> topics);
+        Task UnsubscribeAsync(IEnumerable<string> topics);
     }
 }

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -288,7 +288,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        public Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters)
+        public Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters)
         {
             ThrowIfDisposed();
 
@@ -316,7 +316,7 @@ namespace MQTTnet.Extensions.ManagedClient
             return CompletedTask.Instance;
         }
 
-        public Task UnsubscribeAsync(ICollection<string> topics)
+        public Task UnsubscribeAsync(IEnumerable<string> topics)
         {
             ThrowIfDisposed();
 

Source/MQTTnet.Extensions.TopicTemplate/MQTTnet.Extensions.TopicTemplate.csproj

@@ -11,10 +11,10 @@
         <Company>The contributors of MQTTnet</Company>
         <Product>MQTTnet</Product>
         <Description>
-           This is an extension library which provides mqtt topic templating logic to support dispatch,
-           routing and similar functionality based on the well known moustache syntax, aka
-           AsyncAPI dynamic channel address.
+           Provides mqtt topic templating logic to support dispatch,
+           routing and similar functionality based on the well known moustache syntax (AsyncAPI compatible).
         </Description>
+		<PackageReadmeFile>README.md</PackageReadmeFile>
         <Authors>The contributors of MQTTnet</Authors>
         <PackageId>MQTTnet.Extensions.TopicTemplate</PackageId>
         <SignAssembly>false</SignAssembly>
@@ -51,6 +51,10 @@
     </PropertyGroup>
 
     <ItemGroup>
+        <None Include="README.md">
+            <Pack>True</Pack>
+            <PackagePath>\</PackagePath>
+        </None>
         <None Include="..\..\Images\nuget.png">
             <Pack>True</Pack>
             <PackagePath>\</PackagePath>

Source/MQTTnet.Extensions.TopicTemplate/MqttTopicTemplate.cs

@@ -16,8 +16,8 @@ namespace MQTTnet.Extensions.TopicTemplate
     ///     segments in curly braces called parameters. This well-known
     ///     'moustache' syntax also matches AsyncAPI Channel Address Expressions.
     ///     The topic template is designed to support dynamic subscription/publication,
-    ///     message-topic matching and routing. It is intended to be more convenient
-    ///     than String.Format() for aforementioned purposes.
+    ///     message-topic matching and routing. It is intended to be more safe and
+    ///     convenient than String.Format() for aforementioned purposes.
     /// </summary>
     /// <example>
     ///     topic/subtopic/{parameter}/{otherParameter}
@@ -282,12 +282,17 @@ namespace MQTTnet.Extensions.TopicTemplate
         /// <returns>the topic template (without the parameter)</returns>
         public MqttTopicTemplate WithoutParameter(string parameter)
         {
-            return WithParameter(parameter, MqttTopicFilterComparer.SingleLevelWildcard.ToString());
+            if (string.IsNullOrEmpty(parameter) || !_parameterSegments.Contains(parameter))
+            {
+                throw new ArgumentException("topic template parameter must exist.");
+            }
+            
+            return ReplaceInternal(parameter, MqttTopicFilterComparer.SingleLevelWildcard.ToString());
         }
         
         /// <summary>
         ///     Substitute a parameter with a given value, thus removing the parameter. If the parameter is not present,
-        ///     the method trows. The value must not contain slashes.
+        ///     the method trows. The value must not contain slashes or wildcards.
         /// </summary>
         /// <param
         ///     name="parameter">
@@ -304,12 +309,19 @@ namespace MQTTnet.Extensions.TopicTemplate
         /// <returns>the topic template (without the parameter)</returns>
         public MqttTopicTemplate WithParameter(string parameter, string value)
         {
-            if (value == null || string.IsNullOrEmpty(parameter) || !_parameterSegments.Contains(parameter) || value.Contains(MqttTopicFilterComparer.LevelSeparator) ||
+            if (value == null || string.IsNullOrEmpty(parameter) || !_parameterSegments.Contains(parameter) ||
+                value.Contains(MqttTopicFilterComparer.LevelSeparator) ||
+                value.Contains(MqttTopicFilterComparer.SingleLevelWildcard) ||
                 value.Contains(MqttTopicFilterComparer.MultiLevelWildcard))
             {
-                throw new ArgumentException("parameter must exist and value must not contain slashes.");
+                throw new ArgumentException("parameter must exist and value must not contain slashes or wildcard.");
             }
             
+            return ReplaceInternal(parameter, value);
+        }
+        
+        private MqttTopicTemplate ReplaceInternal(string parameter, string value)
+        {
             var moustache = "{" + parameter + "}";
             return new MqttTopicTemplate(Template.Replace(moustache, value));
         }

Source/MQTTnet.Extensions.TopicTemplate/README.md

@@ -0,0 +1,79 @@
+# MQTTNet.Extensions.TopicTemplate
+
+Provides mqtt topic templating logic to support dispatch, routing and
+similar functionality based on the well known moustache syntax, aka
+AsyncAPI dynamic channel address.
+
+Generating and parsing MQTT topics and topic filters is often done in
+an ad-hoc fashion, leading to error-prone code and subtly introduced
+showstoppers like forgetting to strip a slash from a parameter.  To
+remedy, this extension aids you write safe, maintainable logic
+for topic generation, routing and dispatch.
+
+
+## How it works
+
+The package lets you write MQTTNet code like:
+```csharp
+var template = new MqttTopicTemplate("App/v1/{sender}/message");
+
+var filter = new MqttTopicFilterBuilder()
+    .WithTopicTemplate(template) // (1)
+    .WithAtLeastOnceQoS()
+    .WithNoLocal()
+    .Build();
+
+// filter.Topic == "App/v1/+/message"
+
+var myTopic = template
+    .WithParameter("sender", "me, myself & i"); // (2)
+
+var message = new MqttApplicationMessageBuilder()
+    .WithTopicTemplate(  // (3)
+        myTopic)
+    .WithPayload("Hello!")
+    .Build();
+
+// message.Topic == "App/v1/me, myself & i/message"
+
+Assert.IsTrue(message.MatchesTopicTemplate(template)); // (4)
+```
+
+At (1), the parameter "sender" was automatically replaced by a plus sign
+(single level wildcard) because the caller is creating a topic filter.
+
+Using the same template (they are immutable), you can send
+a message. At (2), we replace the parameter with a known value.
+If that value contained a level separator or topic filter only
+characters, we'd get an ArgumentException. If we forgot to replace
+the parameter "sender" (e.g. when placing template not myTopic),
+we would be prevented from sending at (3).
+
+On the receiving side, a message can be matched to a template
+using code like at (4).
+
+These daily tasks become much safer and easier to code than 
+string.Format() and Split() allow for. More elaborate computation is supported
+too. You could extract the sender using the template or route
+messages based on topic content.
+
+## Features
+
+- Safe handling of topic parameters in moustache syntax
+- Simplifies topic generation and parsing logic
+- Match topics and extract the values of their parameters
+- Translate between topic templates to ease dynamic routing
+- Derive a canonical topic filter from a bunch of templates
+- Integrates with MQTTNet message, subscription and filter builders
+- Extension method friendly (sealed class with one public ctor)
+
+## Additional documentation
+
+- Some of the [unit tests](https://github.com/dotnet/MQTTnet/blob/master/Source/MQTTnet.Tests/Extensions/MqttTopicTemplate_Tests.cs) are instructive
+- Some examples feature the extension, e.g. TopicGenerator, Client_Subscribe_Samples
+
+## Feedback
+
+Report any issues with the main project on https://github.com/dotnet/MQTTnet/issues
+
+

Source/MQTTnet.Extensions.TopicTemplate/TopicTemplateExtensions.cs

@@ -4,6 +4,8 @@
 
 using System;
 using System.Linq;
+using MQTTnet.Client;
+using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.TopicTemplate
@@ -55,6 +57,24 @@ namespace MQTTnet.Extensions.TopicTemplate
             return new MqttTopicFilterBuilder().WithTopicTemplate(topicTemplate, subscribeTreeRoot);
         }
 
+        /// <summary>
+        ///     Create a message builder from this template. The template must not have
+        ///     remaining parameters. 
+        /// </summary>
+        /// <param
+        ///     name="topicTemplate">
+        ///     a parameterless topic template
+        /// </param>
+        /// <returns>a new message builder</returns>
+        /// <exception
+        ///     cref="ArgumentException">
+        ///     if the topic template has parameters
+        /// </exception>
+        public static MqttApplicationMessageBuilder BuildMessage(this MqttTopicTemplate topicTemplate)
+        {
+            return new MqttApplicationMessageBuilder().WithTopicTemplate(topicTemplate);
+        }
+        
         /// <summary>
         ///     Return a message builder to respond to this message. The
         ///     message's response topic and correlation data are included
@@ -116,6 +136,29 @@ namespace MQTTnet.Extensions.TopicTemplate
             return builder.WithTopic(subscribeTreeRoot ? topicTemplate.TopicTreeRootFilter : topicTemplate.TopicFilter);
         }
 
+        /// <summary>
+        /// Set the subscription to the template's topic filter.
+        /// </summary>
+        /// <returns>the builder</returns>
+        public static MqttClientSubscribeOptionsBuilder WithTopicTemplate(
+            this MqttClientSubscribeOptionsBuilder builder,
+            MqttTopicTemplate topicTemplate,
+            MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
+            bool noLocal = false,
+            bool retainAsPublished = false,
+            MqttRetainHandling retainHandling = MqttRetainHandling.SendAtSubscribe)
+        {
+            return builder.WithTopicFilter(
+                new MqttTopicFilter
+                {
+                    Topic = topicTemplate.TopicFilter,
+                    QualityOfServiceLevel = qualityOfServiceLevel,
+                    NoLocal = noLocal,
+                    RetainAsPublished = retainAsPublished,
+                    RetainHandling = retainHandling
+                });
+        }
+        
         /// <summary>
         ///     Set the publication topic according to the topic template. The template
         ///     must not have remaining (unset) parameters or contain wildcards.

Source/MQTTnet.TestApp/MQTTnet.TestApp.csproj

@@ -21,6 +21,7 @@
         <ProjectReference Include="..\..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj" />
         <ProjectReference Include="..\..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj" />
         <ProjectReference Include="..\..\Source\MQTTnet\MQTTnet.csproj" />
+        <ProjectReference Include="..\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj" />
     </ItemGroup>
 
 </Project>

Source/MQTTnet.TestApp/TopicGenerator.cs

@@ -1,6 +1,6 @@
 using System;
 using System.Collections.Generic;
-using System.Text;
+using MQTTnet.Extensions.TopicTemplate;
 
 namespace MQTTnet.TestApp
 {
@@ -17,6 +17,8 @@ namespace MQTTnet.TestApp
             singleWildcardTopicsByPublisher = new Dictionary<string, List<string>>();
             multiWildcardTopicsByPublisher = new Dictionary<string, List<string>>();
 
+            MqttTopicTemplate baseTemplate = new MqttTopicTemplate("{publisher}/{building}/{level}/{sensor}");
+
             // Find some reasonable distribution across three topic levels
             var topicsPerLevel = (int)Math.Pow(numTopicsPerPublisher, (1.0 / 3.0));
 
@@ -40,25 +42,37 @@ namespace MQTTnet.TestApp
                 int publisherTopicCount = 0;
 
                 var publisherName = "pub" + p;
+                var publisherTemplate = baseTemplate
+                    .WithParameter("publisher", publisherName);
                 for (var l1 = 0; l1 < numLevel1Topics; ++l1)
                 {
+                    var l1Template = publisherTemplate
+                        .WithParameter("building", "building" + (l1 + 1));
                     for (var l2 = 0; l2 < numLevel2Topics; ++l2)
                     {
+                        var l2Template = l1Template
+                            .WithParameter("level", "level" + (l2 + 1));
                         for (var l3 = 0; l3 < maxNumLevel3Topics; ++l3)
                         {
                             if (publisherTopicCount >= numTopicsPerPublisher)
                                 break;
 
-                            var topic = string.Format("{0}/building{1}/level{2}/sensor{3}", publisherName, l1 + 1, l2 + 1, l3 + 1);
+                            var topic = l2Template
+                                .WithParameter("sensor", "sensor" + (l3 + 1))
+                                .TopicFilter;
                             AddPublisherTopic(publisherName, topic, topicsByPublisher);
 
                             if (l2 == 0)
                             {
-                                var singleWildcardTopic = string.Format("{0}/building{1}/+/sensor{2}", publisherName, l1 + 1, l3 + 1);
+                                var singleWildcardTopic = l1Template
+                                        .WithParameter("sensor", "sensor" + (l3 + 1))
+                                        .TopicFilter;
                                 AddPublisherTopic(publisherName, singleWildcardTopic, singleWildcardTopicsByPublisher);
                                 if (l1 == 0)
                                 {
-                                    var multiWildcardTopic = string.Format("{0}/+/level{1}/+", publisherName, l3 + 1);
+                                    var multiWildcardTopic = publisherTemplate
+                                        .WithParameter("sensor", "sensor" + (l3 + 1))
+                                        .TopicFilter;
                                     AddPublisherTopic(publisherName, multiWildcardTopic, multiWildcardTopicsByPublisher);
                                 }
                             }

Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -167,6 +168,91 @@ namespace MQTTnet.Tests.Clients.MqttClient
             }
         }
 
+        class TestClientKerberosAuthenticationHandler : IMqttExtendedAuthenticationExchangeHandler
+        {
+            public static IReadOnlyList<string> ServerTokens { get; } = new List<string>()
+            {
+                string.Empty,
+                "reply context token"
+            };
+
+            public static IReadOnlyList<string> ClientTokens { get; } = new List<string>()
+            {
+                string.Empty,
+                "initial context token",
+                string.Empty
+            };
+
+            int tokenCursor;
+
+            public async Task HandleRequestAsync(MqttExtendedAuthenticationExchangeContext context)
+            {
+                if (context.AuthenticationMethod != "GS2-KRB5")
+                {
+                    throw new InvalidOperationException("Wrong authentication method");
+                }
+
+                Assert.IsTrue(tokenCursor < ServerTokens.Count, "No more server tokens were expected.");
+                var expectedToken = ServerTokens[tokenCursor++];
+                Assert.AreEqual(expectedToken, Encoding.UTF8.GetString(context.AuthenticationData));
+
+                var sendOptions = new MqttExtendedAuthenticationExchangeData
+                {
+                    AuthenticationData = Encoding.UTF8.GetBytes(ClientTokens[tokenCursor])
+                };
+
+                await context.Client.SendExtendedAuthenticationExchangeDataAsync(sendOptions, context.CancellationToken);
+            }
+        }
+
+        [TestMethod]
+        public async Task Use_Extended_Authentication()
+        {
+            // Use Kerberos sample from the MQTT RFC.
+            var kerberosAuthenticationHandler = new TestClientKerberosAuthenticationHandler();
+
+            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
+            {
+                var server = await testEnvironment.StartServer();
+
+                server.ValidatingConnectionAsync += async args =>
+                {
+                    if (args.AuthenticationMethod == "GS2-KRB5")
+                    {
+                        var authenticationData = args.AuthenticationData;
+                        string expectedToken;
+                        int tokenCursor;
+                        for (tokenCursor = 0; tokenCursor < TestClientKerberosAuthenticationHandler.ServerTokens.Count; tokenCursor++)
+                        {
+                            expectedToken = TestClientKerberosAuthenticationHandler.ClientTokens[tokenCursor];
+                            Assert.AreEqual(expectedToken, Encoding.UTF8.GetString(authenticationData), "The received client token is not correct.");
+
+                            var serverToken = TestClientKerberosAuthenticationHandler.ServerTokens[tokenCursor];
+                            var authOptions = testEnvironment.Factory.CreateMqttExtendedAuthenticationExchangeOptionsBuilder().WithAuthenticationData(serverToken).Build();
+                            var result = await args.SendExtendedAuthenticationExchangeDataAsync(authOptions);
+
+                            authenticationData = result.AuthenticationData;
+                        }
+
+                        expectedToken = TestClientKerberosAuthenticationHandler.ClientTokens[tokenCursor];
+                        Assert.AreEqual(expectedToken, Encoding.UTF8.GetString(authenticationData), "The received client token is not correct.");
+                        args.ResponseAuthenticationData = Encoding.UTF8.GetBytes("outcome of authentication");
+                        args.ReasonCode = MqttConnectReasonCode.Success;
+                        args.ReasonString = "Authentication successful";
+                    }
+                    else
+                    {
+                        args.ReasonCode = MqttConnectReasonCode.BadAuthenticationMethod;
+                    }
+                };
+
+                var clientOptions = testEnvironment.CreateDefaultClientOptionsBuilder().WithAuthentication("GS2-KRB5", new byte[0]).WithExtendedAuthenticationExchangeHandler(kerberosAuthenticationHandler);
+                var client = await testEnvironment.ConnectClient(clientOptions);
+
+                Assert.IsTrue(client.IsConnected);
+            }
+        }
+
         [TestMethod]
         public async Task No_Unobserved_Exception()
         {

Source/MQTTnet.Tests/Extensions/MqttTopicTemplate_Tests.cs

@@ -45,6 +45,29 @@ namespace MQTTnet.Tests.Extensions
             var template = new MqttTopicTemplate("A/B/{foo}/D");
             template.WithParameter("foo", "a#");
         }
+
+        [TestMethod]
+        [ExpectedException(typeof(ArgumentException))]
+        public void RejectsReservedChars2()
+        {
+            var template = new MqttTopicTemplate("A/B/{foo}/D");
+            template.WithParameter("foo", "a+b");
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(ArgumentException))]
+        public void RejectsReservedChars3()
+        {
+            var template = new MqttTopicTemplate("A/B/{foo}/D");
+            template.WithParameter("foo", "a/b");
+        }
+        
+        [TestMethod]
+        public void AcceptsEmptyValue()
+        {
+            var template = new MqttTopicTemplate("A/B/{foo}/D");
+            template.WithParameter("foo", "");
+        }
         
         [TestMethod]
         [ExpectedException(typeof(MqttProtocolViolationException))]
@@ -60,13 +83,6 @@ namespace MQTTnet.Tests.Extensions
             var _ = new MqttTopicTemplate(null);
         }
 
-        [TestMethod]
-        [ExpectedException(typeof(ArgumentException))]
-        public void RejectsReservedChars2()
-        {
-            var template = new MqttTopicTemplate("A/B/{foo}/D");
-            template.WithParameter("foo", "e/f");
-        }
         
         [TestMethod]
         public void IgnoresEmptyParameters()
@@ -114,6 +130,50 @@ namespace MQTTnet.Tests.Extensions
             Assert.AreEqual("A/v1/+/F", filter.Topic);
         }
 
+        [TestMethod]
+        public void SubscriptionSupport2()
+        {
+            var template = new MqttTopicTemplate("A/v1/{param}/F");
+            
+            var subscribeOptions = new MqttFactory().CreateSubscribeOptionsBuilder()
+                .WithTopicTemplate(template)
+                .WithSubscriptionIdentifier(5)
+                .Build();
+            Assert.AreEqual("A/v1/+/F", subscribeOptions.TopicFilters[0].Topic);
+        }
+
+        [TestMethod]
+        public void SendAndSubscribeSupport()
+        {
+            var template = new MqttTopicTemplate("App/v1/{sender}/message");
+
+            var filter = new MqttTopicFilterBuilder()
+                .WithTopicTemplate(template)
+                .WithAtLeastOnceQoS()
+                .WithNoLocal()
+                .Build();
+
+            Assert.AreEqual("App/v1/+/message", filter.Topic);
+
+            var myTopic = template.WithParameter("sender", "me, myself & i");
+
+            var message = new MqttApplicationMessageBuilder()
+                .WithTopicTemplate(
+                    myTopic)
+                .WithPayload("Hello!")
+                .Build();
+
+            Assert.IsTrue(message.MatchesTopicTemplate(template));
+        }
+
+        [TestMethod]
+        public void SendAndSubscribeSupport2()
+        {
+            var template = new MqttTopicTemplate("App/v1/{sender}/message");
+            Assert.ThrowsException<ArgumentException>(() => 
+                template.BuildMessage());
+        }
+
         [TestMethod]
         public void CanonicalPrefixFilter()
         {

Source/MQTTnet.Tests/MqttTcpChannel_Tests.cs

@@ -2,13 +2,13 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using Microsoft.VisualStudio.TestTools.UnitTesting;
-using MQTTnet.Implementations;
 using System;
 using System.Net;
 using System.Net.Sockets;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Implementations;
 
 namespace MQTTnet.Tests
 {
@@ -18,62 +18,68 @@ namespace MQTTnet.Tests
         [TestMethod]
         public async Task Dispose_Channel_While_Used()
         {
-            var ct = new CancellationTokenSource();
-            var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
-
-            try
+            using (var ct = new CancellationTokenSource())
             {
-                serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001));
-                serverSocket.Listen(0);
-
-#pragma warning disable 4014
-                Task.Run(async () =>
-#pragma warning restore 4014
+                using (var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp))
                 {
-                    while (!ct.IsCancellationRequested)
+                    try
                     {
-                        var client = await serverSocket.AcceptAsync();
-                        var data = new byte[] { 128 };
-                        await client.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
-                    }
-                }, ct.Token);
+                        serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001));
+                        serverSocket.Listen(0);
 
-                var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
-                await clientSocket.ConnectAsync("localhost", 50001, CancellationToken.None);
-
-                var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), "test", null);
-
-                await Task.Delay(100, ct.Token);
-
-                var buffer = new byte[1];
-                await tcpChannel.ReadAsync(buffer, 0, 1, ct.Token);
-
-                Assert.AreEqual(128, buffer[0]);
-
-                // This block should fail after dispose.
 #pragma warning disable 4014
-                Task.Run(() =>
+                        Task.Run(
+                            async () =>
 #pragma warning restore 4014
-                {
-                    Task.Delay(200, ct.Token);
-                    tcpChannel.Dispose();
-                }, ct.Token);
+                            {
+                                while (!ct.IsCancellationRequested)
+                                {
+                                    var client = await serverSocket.AcceptAsync(ct.Token);
+                                    var data = new byte[] { 128 };
+                                    await client.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
+                                }
+                            },
+                            ct.Token);
 
-                try
-                {
-                    await tcpChannel.ReadAsync(buffer, 0, 1, CancellationToken.None);
+                        var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
+                        await clientSocket.ConnectAsync("localhost", 50001, CancellationToken.None);
+
+                        var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), "test", null);
+
+                        await Task.Delay(100, ct.Token);
+
+                        var buffer = new byte[1];
+                        await tcpChannel.ReadAsync(buffer, 0, 1, ct.Token);
+
+                        Assert.AreEqual(128, buffer[0]);
+
+                        // This block should fail after dispose.
+#pragma warning disable 4014
+                        Task.Run(
+                            () =>
+#pragma warning restore 4014
+                            {
+                                Task.Delay(200, ct.Token);
+                                tcpChannel.Dispose();
+                            },
+                            ct.Token);
+
+                        try
+                        {
+                            await tcpChannel.ReadAsync(buffer, 0, 1, CancellationToken.None);
+                        }
+                        catch (Exception exception)
+                        {
+                            Assert.IsInstanceOfType(exception, typeof(SocketException));
+                            Assert.AreEqual(SocketError.OperationAborted, ((SocketException)exception).SocketErrorCode);
+                        }
+                    }
+                    finally
+                    {
+                        ct.Cancel(false);
+                    }
                 }
-                catch (Exception exception)
-                {
-                    Assert.IsInstanceOfType(exception, typeof(SocketException));
-                    Assert.AreEqual(SocketError.OperationAborted, ((SocketException)exception).SocketErrorCode);
-                }
-            }
-            finally
-            {
-                ct.Cancel(false);
-                serverSocket.Dispose();
             }
         }
     }
-}
+}

Source/MQTTnet.Tests/TopicGenerator.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using MQTTnet.Extensions.TopicTemplate;
 
 namespace MQTTnet.Tests
 {
@@ -31,31 +32,43 @@ namespace MQTTnet.Tests
             {
                 maxNumLevel3Topics = 1;
             }
+            
+            MqttTopicTemplate baseTemplate = new MqttTopicTemplate("{publisher}/{building}/{level}/{sensor}");
 
             for (var p = 0; p < numPublishers; ++p)
             {
                 int publisherTopicCount = 0;
                 var publisherName = "pub" + p;
+                var publisherTemplate = baseTemplate
+                    .WithParameter("publisher", publisherName);
+                
                 for (var l1 = 0; l1 < numLevel1Topics; ++l1)
                 {
+                    var l1Template = publisherTemplate.WithParameter("building", "building" + (l1 + 1));
                     for (var l2 = 0; l2 < numLevel2Topics; ++l2)
                     {
+                        var l2Template = l1Template.WithParameter("level", "level" + (l2 + 1));
                         for (var l3 = 0; l3 < maxNumLevel3Topics; ++l3)
                         {
                             if (publisherTopicCount >= numTopicsPerPublisher)
                                 break;
 
-                            var topic = string.Format("{0}/building{1}/level{2}/sensor{3}", publisherName, l1 + 1, l2 + 1, l3 + 1);
-                            AddPublisherTopic(publisherName, topic, topicsByPublisher);
+                            var l3Template = l2Template
+                                .WithParameter("sensor", "sensor" + (l3 + 1));
+                            AddPublisherTopic(publisherName, l3Template.TopicFilter, topicsByPublisher);
 
                             if (l2 == 0)
                             {
-                                var singleWildcardTopic = string.Format("{0}/building{1}/+/sensor{2}", publisherName, l1 + 1, l3 + 1);
+                                var singleWildcardTopic = l1Template
+                                    .WithParameter("sensor", "sensor" + (l3 + 1))
+                                    .TopicFilter;
                                 AddPublisherTopic(publisherName, singleWildcardTopic, singleWildcardTopicsByPublisher);
                             }
                             if ((l1 == 0) && (l3 == 0))
                             {
-                                var multiWildcardTopic = string.Format("{0}/+/level{1}/+", publisherName, l2 + 1);
+                                var multiWildcardTopic = publisherTemplate
+                                    .WithParameter("level", "level" + (l2 + 1))
+                                    .TopicFilter;
                                 AddPublisherTopic(publisherName, multiWildcardTopic, multiWildcardTopicsByPublisher);
                             }
 

Source/MQTTnet/Client/ExtendedAuthenticationExchange/MqttExtendedAuthenticationExchangeContext.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 
@@ -11,7 +12,7 @@ namespace MQTTnet.Client
 {
     public class MqttExtendedAuthenticationExchangeContext
     {
-        public MqttExtendedAuthenticationExchangeContext(MqttAuthPacket authPacket, MqttClient client)
+        public MqttExtendedAuthenticationExchangeContext(MqttAuthPacket authPacket, MqttClient client, CancellationToken cancellationToken)
         {
             if (authPacket == null) throw new ArgumentNullException(nameof(authPacket));
 
@@ -22,6 +23,7 @@ namespace MQTTnet.Client
             UserProperties = authPacket.UserProperties;
 
             Client = client ?? throw new ArgumentNullException(nameof(client));
+            CancellationToken = cancellationToken;
         }
 
         /// <summary>
@@ -58,5 +60,10 @@ namespace MQTTnet.Client
         public List<MqttUserProperty> UserProperties { get; }
 
         public MqttClient Client { get; }
+
+        /// <summary>
+        /// The cancellation token passed to the operation being authenticated (e.g. ConnectAsync).
+        /// </summary>
+        public CancellationToken CancellationToken { get; }
     }
 }

Source/MQTTnet/Client/MqttClient.cs

@@ -89,6 +89,8 @@ namespace MQTTnet.Client
 
         public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected;
 
+        bool IsConnecting => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connecting;
+
         public MqttClientOptions Options { get; private set; }
 
         public async Task<MqttClientConnectResult> ConnectAsync(MqttClientOptions options, CancellationToken cancellationToken = default)
@@ -304,7 +306,7 @@ namespace MQTTnet.Client
             }
 
             ThrowIfDisposed();
-            ThrowIfNotConnected();
+            ThrowIfNotConnectedOrConnecting();
 
             var authPacket = new MqttAuthPacket
             {
@@ -454,27 +456,30 @@ namespace MQTTnet.Client
                 var connectPacket = MqttPacketFactories.Connect.Create(options);
                 await Send(connectPacket, cancellationToken).ConfigureAwait(false);
 
-                var receivedPacket = await Receive(cancellationToken).ConfigureAwait(false);
-
-                switch (receivedPacket)
+                for (; ; )
                 {
-                    case MqttConnAckPacket connAckPacket:
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    var receivedPacket = await Receive(cancellationToken).ConfigureAwait(false);
+
+                    if (receivedPacket is MqttAuthPacket authPacket)
+                    {
+                        await ProcessReceivedAuthPacket(authPacket, cancellationToken).ConfigureAwait(false);
+                        continue;
+                    }
+
+                    if (receivedPacket is MqttConnAckPacket connAckPacket)
                     {
                         result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, channelAdapter.PacketFormatterAdapter.ProtocolVersion);
                         break;
                     }
-                    case MqttAuthPacket _:
+
+                    if (receivedPacket != null)
                     {
-                        throw new NotSupportedException("Extended authentication handler is not yet supported");
-                    }
-                    case null:
-                    {
-                        throw new MqttCommunicationException("Connection closed.");
-                    }
-                    default:
-                    {
-                        throw new InvalidOperationException($"Received an unexpected MQTT packet ({receivedPacket}).");
+                        throw new MqttProtocolViolationException($"Received other packet than CONNACK or AUTH while connecting ({receivedPacket}).");
                     }
+
+                    throw new MqttCommunicationException("Connection closed.");
                 }
             }
             catch (Exception exception)
@@ -678,10 +683,37 @@ namespace MQTTnet.Client
             return _events.ConnectedEvent.InvokeAsync(eventArgs);
         }
 
-        Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket)
+        Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket, CancellationToken cancellationToken)
         {
             var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler;
-            return extendedAuthenticationExchangeHandler != null ? extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this)) : CompletedTask.Instance;
+            if (extendedAuthenticationExchangeHandler == null)
+            {
+                // From RFC:
+                // The Client can close the connection at any point in this process. It MAY send a DISCONNECT packet before doing so.
+                // The Server can reject the authentication at any point in this process. It MAY send a CONNACK with a Reason Code of
+                // 0x80 or above as described in section 4.13, and MUST close the Network Connection [MQTT-4.12.0-4].
+                //
+                // If the re-authentication fails, the Client or Server SHOULD send DISCONNECT with an appropriate Reason Code
+                // as described in section 4.13, and MUST close the Network Connection [MQTT-4.12.1-2].
+                //
+                // Since we have no handler there is no chance to fulfil the initial authentication or re-authentication requests.
+
+                if (IsConnecting)
+                {
+                    throw new MqttCommunicationException("No handler was provided to handle the extended authentication exchange.");
+                }
+
+                _ = DisconnectAsync(new MqttClientDisconnectOptions
+                {
+                    Reason = MqttClientDisconnectOptionsReason.ImplementationSpecificError,
+                    ReasonString = "Unable to handle AUTH packet"
+                },
+                cancellationToken);
+
+                return CompletedTask.Instance;
+            }
+
+            return extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this, cancellationToken));
         }
 
         Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket)
@@ -935,6 +967,14 @@ namespace MQTTnet.Client
             }
         }
 
+        void ThrowIfNotConnectedOrConnecting()
+        {
+            if (!(IsConnected || IsConnecting))
+            {
+                ThrowNotConnectedOrConnecting();
+            }
+        }
+
         static void ThrowIfOptionsInvalid(MqttClientOptions options)
         {
             if (options == null)
@@ -958,6 +998,11 @@ namespace MQTTnet.Client
             throw new MqttClientNotConnectedException();
         }
 
+        static void ThrowNotConnectedOrConnecting()
+        {
+            throw new MqttCommunicationException("The MQTT client is not connected or connecting.");
+        }
+
         void TryInitiateDisconnect()
         {
             lock (_disconnectLock)
@@ -992,7 +1037,7 @@ namespace MQTTnet.Client
                         await ProcessReceivedDisconnectPacket(disconnectPacket).ConfigureAwait(false);
                         break;
                     case MqttAuthPacket authPacket:
-                        await ProcessReceivedAuthPacket(authPacket).ConfigureAwait(false);
+                        await ProcessReceivedAuthPacket(authPacket, cancellationToken).ConfigureAwait(false);
                         break;
                     case MqttPingRespPacket _:
                         _packetDispatcher.TryDispatch(packet);
@@ -1051,10 +1096,13 @@ namespace MQTTnet.Client
 
                     if (timeWithoutPacketSent > keepAlivePeriod)
                     {
-                        using (var timeoutCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
+                        using (var pingTimeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
                         {
-                            timeoutCancellationTokenSource.CancelAfter(Options.Timeout);
-                            await PingAsync(timeoutCancellationTokenSource.Token).ConfigureAwait(false);
+                            // We already reached the keep alive timeout. Due to the RFC part [MQTT-3.1.2-24] the server will wait another
+                            // 1/2 of the keep alive time. So we can also use this value as the timeout.
+                            pingTimeout.CancelAfter((int)(keepAlivePeriod.TotalMilliseconds / 2));
+
+                            await PingAsync(pingTimeout.Token).ConfigureAwait(false);
                         }
                     }
 

Source/MQTTnet/Client/Options/MqttClientOptionsValidator.cs

@@ -68,8 +68,13 @@ namespace MQTTnet.Client
                 Throw(nameof(options.AuthenticationMethod));
             }
 
+            if (options.ExtendedAuthenticationExchangeHandler != null)
+            {
+                Throw(nameof(options.ExtendedAuthenticationExchangeHandler));
+            }
+
             // Will relevant properties.
-            
+
             if (options.WillPayloadFormatIndicator != MqttPayloadFormatIndicator.Unspecified)
             {
                 Throw(nameof(options.WillPayloadFormatIndicator));

Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs

@@ -68,12 +68,12 @@ namespace MQTTnet.Formatter.V5
 
         MqttPacket DecodeAuthPacket(ArraySegment<byte> body)
         {
-            ThrowIfBodyIsEmpty(body);
-
             _bufferReader.SetBuffer(body.Array, body.Offset, body.Count);
 
             var packet = new MqttAuthPacket();
 
+            // MQTT spec: The Reason Code and Property Length can be omitted if the Reason Code is 0x00 (Success) and there are no Properties.
+            // In this case the AUTH has a Remaining Length of 0.
             if (_bufferReader.EndOfStream)
             {
                 packet.ReasonCode = MqttAuthenticateReasonCode.Success;

Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs

@@ -13,7 +13,7 @@ namespace MQTTnet.Formatter.V5
     public sealed class MqttV5PacketEncoder
     {
         const int FixedHeaderSize = 1;
-        
+
         readonly MqttBufferWriter _bufferWriter;
         readonly MqttV5PropertiesWriter _propertiesWriter = new MqttV5PropertiesWriter(new MqttBufferWriter(1024, 4096));
 
@@ -64,13 +64,20 @@ namespace MQTTnet.Formatter.V5
 
         byte EncodeAuthPacket(MqttAuthPacket packet)
         {
-            _bufferWriter.WriteByte((byte)packet.ReasonCode);
-
             _propertiesWriter.WriteAuthenticationMethod(packet.AuthenticationMethod);
             _propertiesWriter.WriteAuthenticationData(packet.AuthenticationData);
             _propertiesWriter.WriteReasonString(packet.ReasonString);
             _propertiesWriter.WriteUserProperties(packet.UserProperties);
 
+            // MQTT spec: The Reason Code and Property Length can be omitted if the Reason Code is 0x00 (Success) and there are no Properties.
+            // In this case the AUTH has a Remaining Length of 0.
+            if (packet.ReasonCode == MqttAuthenticateReasonCode.Success && _propertiesWriter.Length == 0)
+            {
+                return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Auth);
+            }
+
+            _bufferWriter.WriteByte((byte)packet.ReasonCode);
+
             _propertiesWriter.WriteTo(_bufferWriter);
             _propertiesWriter.Reset();
 
@@ -526,7 +533,7 @@ namespace MQTTnet.Formatter.V5
 
         static void ThrowIfPacketIdentifierIsInvalid(ushort packetIdentifier, MqttPacket packet)
         {
-            // SUBSCRIBE, UNSUBSCRIBE, and PUBLISH(in cases where QoS > 0) Control Packets MUST contain a non-zero 16 - bit Packet Identifier[MQTT - 2.3.1 - 1]. 
+            // SUBSCRIBE, UNSUBSCRIBE, and PUBLISH(in cases where QoS > 0) Control Packets MUST contain a non-zero 16 - bit Packet Identifier[MQTT - 2.3.1 - 1].
 
             if (packetIdentifier == 0)
             {

Source/MQTTnet/Implementations/CrossPlatformSocket.cs

@@ -142,12 +142,14 @@ namespace MQTTnet.Implementations
             set => _socket.SendTimeout = value;
         }
 
-        public async Task<CrossPlatformSocket> AcceptAsync()
+        public async Task<CrossPlatformSocket> AcceptAsync(CancellationToken cancellationToken)
         {
             try
             {
 #if NET452 || NET461
                 var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false);
+#elif NET7_0_OR_GREATER
+                var clientSocket = await _socket.AcceptAsync(cancellationToken).ConfigureAwait(false);
 #else
                 var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
 #endif

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -141,7 +141,7 @@ namespace MQTTnet.Implementations
             {
                 try
                 {
-                    var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
+                    var clientSocket = await _socket.AcceptAsync(cancellationToken).ConfigureAwait(false);
                     if (clientSocket == null)
                     {
                         continue;

Source/MQTTnet/MqttFactory.cs

@@ -135,6 +135,11 @@ namespace MQTTnet
             return new MqttClient(clientAdapterFactory, logger);
         }
 
+        public MqttExtendedAuthenticationExchangeOptionsFactory CreateMqttExtendedAuthenticationExchangeOptionsBuilder()
+        {
+            return new MqttExtendedAuthenticationExchangeOptionsFactory();
+        }
+
         public MqttServer CreateMqttServer(MqttServerOptions options)
         {
             return CreateMqttServer(options, DefaultLogger);

Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs

@@ -7,11 +7,14 @@ using System.Collections;
 using System.Collections.Generic;
 using System.Security.Cryptography.X509Certificates;
 using System.Text;
+using System.Threading.Tasks;
+using System.Threading;
 using MQTTnet.Adapter;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Exceptions;
 
 namespace MQTTnet.Server
 {
@@ -189,5 +192,54 @@ namespace MQTTnet.Server
         ///     A value of 0 indicates that the value is not used.
         /// </summary>
         public uint WillDelayInterval => _connectPacket.WillDelayInterval;
+
+
+        public async Task<MqttExtendedAuthenticationExchangeResult> SendExtendedAuthenticationExchangeDataAsync(
+            MqttExtendedAuthenticationExchangeOptions options,
+            CancellationToken cancellationToken = default)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+
+            var requestAuthPacket = new MqttAuthPacket
+            {
+                // From RFC: If the initial CONNECT packet included an Authentication Method property then all AUTH packets,
+                // and any successful CONNACK packet MUST include an Authentication Method Property with the same value as in the CONNECT packet [MQTT-4.12.0-5].
+                AuthenticationMethod = AuthenticationMethod,
+
+                // The reason code will stay at continue all the time when connecting. The server will respond with the
+                // CONNACK packet when authentication is done!
+                ReasonCode = MqttAuthenticateReasonCode.ContinueAuthentication,
+
+                AuthenticationData = options.AuthenticationData,
+                ReasonString = options.ReasonString,
+                UserProperties = options.UserProperties
+            };
+
+            await ChannelAdapter.SendPacketAsync(requestAuthPacket, cancellationToken).ConfigureAwait(false);
+
+            var responsePacket = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
+
+            if (responsePacket == null)
+            {
+                throw new MqttCommunicationException("The client closed the connection.");
+            }
+
+            if (responsePacket is MqttAuthPacket responseAuthPacket)
+            {
+                if (!string.Equals(AuthenticationMethod, responseAuthPacket.AuthenticationMethod, StringComparison.Ordinal))
+                {
+                    throw new MqttProtocolViolationException("The authentication method cannot change while authenticating the client.");
+                }
+
+                return MqttExtendedAuthenticationExchangeResultFactory.Create(responseAuthPacket);
+            }
+
+            if (responsePacket is MqttDisconnectPacket disconnectPacket)
+            {
+                return MqttExtendedAuthenticationExchangeResultFactory.Create(disconnectPacket);
+            }
+
+            throw new MqttProtocolViolationException("Received other packet than AUTH while authenticating.");
+        }
     }
 }

Source/MQTTnet/Server/ExtendedAuthenticationExchange/MqttExtendedAuthenticationExchangeOptions.cs

@@ -0,0 +1,18 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using MQTTnet.Packets;
+using System.Collections.Generic;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttExtendedAuthenticationExchangeOptions
+    {
+        public byte[] AuthenticationData { get; set; }
+
+        public string ReasonString { get; set; }
+
+        public List<MqttUserProperty> UserProperties { get; set; }
+    }
+}

Source/MQTTnet/Server/ExtendedAuthenticationExchange/MqttExtendedAuthenticationExchangeOptionsFactory.cs

@@ -0,0 +1,73 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Collections.Generic;
+using System.Text;
+using System;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttExtendedAuthenticationExchangeOptionsFactory
+    {
+        readonly MqttExtendedAuthenticationExchangeOptions _options = new MqttExtendedAuthenticationExchangeOptions();
+
+        public MqttExtendedAuthenticationExchangeOptions Build()
+        {
+            return _options;
+        }
+
+        public MqttExtendedAuthenticationExchangeOptionsFactory WithAuthenticationData(byte[] authenticationData)
+        {
+            _options.AuthenticationData = authenticationData;
+
+            return this;
+        }
+
+        public MqttExtendedAuthenticationExchangeOptionsFactory WithAuthenticationData(string authenticationData)
+        {
+            if (authenticationData == null)
+            {
+                _options.AuthenticationData = null;
+            }
+            else
+            {
+                _options.AuthenticationData = Encoding.UTF8.GetBytes(authenticationData);
+            }
+
+            return this;
+        }
+
+        public MqttExtendedAuthenticationExchangeOptionsFactory WithReasonString(string reasonString)
+        {
+            _options.ReasonString = reasonString;
+
+            return this;
+        }
+
+        public MqttExtendedAuthenticationExchangeOptionsFactory WithUserProperties(List<MqttUserProperty> userProperties)
+        {
+            _options.UserProperties = userProperties;
+
+            return this;
+        }
+
+        public MqttExtendedAuthenticationExchangeOptionsFactory WithUserProperty(string name, string value)
+        {
+            if (name == null)
+            {
+                throw new ArgumentNullException(nameof(name));
+            }
+
+            if (_options.UserProperties == null)
+            {
+                _options.UserProperties = new List<MqttUserProperty>();
+            }
+
+            _options.UserProperties.Add(new MqttUserProperty(name, value));
+
+            return this;
+        }
+    }
+}

Source/MQTTnet/Server/ExtendedAuthenticationExchange/MqttExtendedAuthenticationExchangeResult.cs

@@ -0,0 +1,18 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using MQTTnet.Packets;
+using System.Collections.Generic;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttExtendedAuthenticationExchangeResult
+    {
+        public string ReasonString { get; set; }
+
+        public List<MqttUserProperty> UserProperties { get; set; }
+
+        public byte[] AuthenticationData { get; set; }
+    }
+}

Source/MQTTnet/Server/ExtendedAuthenticationExchange/MqttExtendedAuthenticationExchangeResultFactory.cs

@@ -0,0 +1,46 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using MQTTnet.Packets;
+using System;
+
+namespace MQTTnet.Server
+{
+    public static class MqttExtendedAuthenticationExchangeResultFactory
+    {
+        public static MqttExtendedAuthenticationExchangeResult Create(MqttAuthPacket authPacket)
+        {
+            if (authPacket == null)
+            {
+                throw new ArgumentNullException(nameof(authPacket));
+            }
+
+            return new MqttExtendedAuthenticationExchangeResult
+            {
+                AuthenticationData = authPacket.AuthenticationData,
+
+                ReasonString = authPacket.ReasonString,
+                UserProperties = authPacket.UserProperties
+            };
+        }
+
+        public static MqttExtendedAuthenticationExchangeResult Create(MqttDisconnectPacket disconnectPacket)
+        {
+            if (disconnectPacket == null)
+            {
+                throw new ArgumentNullException(nameof(disconnectPacket));
+            }
+
+            return new MqttExtendedAuthenticationExchangeResult
+            {
+                AuthenticationData = null,
+                ReasonString = disconnectPacket.ReasonString,
+                UserProperties = disconnectPacket.UserProperties
+
+                // SessionExpiryInterval makes no sense because the connection is not yet made!
+                // ServerReferences makes no sense when the client initiated a DISCONNECT!
+            };
+        }
+    }
+}