Update MQTTnet.nuspec

Merge remote-tracking branch 'origin/master'
Update docs. Add UnitTests.
2026-10-15 00:15:15 +08:00 · 2019-12-28 14:05:24 +01:00 · 2019-10-03 15:23:42 +02:00 · 2019-10-03 15:23:35 +02:00 · 2019-09-05 10:37:55 +02:00 · 2019-08-18 21:00:40 +02:00
80 changed files with 1483 additions and 300 deletions
@@ -0,0 +1,46 @@
+---
+name: Bug report
+about: Create a report to help us improve.
+title: ''
+labels: 'bug'
+assignees: ''
+
+---
+
+### Describe the bug
+A clear and concise description of what the bug is.
+
+### Which project is your bug related to?
+<!-- Remove the items which don't apply from the following list -->
+- Client
+- ManagedClient
+- MQTTnet.Server standalone
+- Server
+- Generic
+
+### To Reproduce
+Steps to reproduce the behavior:
+1. Using this version of MQTTnet '...'.
+2. Run this code '....'.
+3. With these arguments '....'.
+4. See error.
+
+### Expected behavior
+A clear and concise description of what you expected to happen.
+
+### Screenshots
+If applicable, add screenshots to help explain your problem.
+
+### Additional context / logging
+Add any other context about the problem here.
+Include debugging or logging information here:
+
+```batch
+\\ Put your logging output here.
+```
+### Code example
+ Please provide full code examples below where possible to make it easier for the developers to check your issues.
+ 
+```csharp
+\\ Put your code here.
+```
@@ -0,0 +1,19 @@
+---
+name: Custom issue template
+about: Do you have a question related to the project? Use this template.
+title: ''
+labels: 'question'
+assignees: ''
+
+---
+
+### Describe your question
+A clear and concise description of what you want to know.
+
+### Which project is your question related to?
+<!-- Remove the items which don't apply from the following list -->
+- Client
+- ManagedClient
+- MQTTnet.Server standalone
+- Server
+- Generic
@@ -0,0 +1,29 @@
+---
+name: Feature request
+about: Suggest an idea for this project.
+title: ''
+labels: 'feature-request'
+assignees: ''
+
+---
+
+### Describe the feature request
+Is your feature request related to a problem? Please describe.
+A clear and concise description of what the problem is. Example. I'm am trying to do [...] but [...]
+
+### Which project is your feature request related to?
+<!-- Remove the items which don't apply from the following list -->
+- Client
+- ManagedClient
+- MQTTnet.Server standalone
+- Server
+- Generic
+
+### Describe the solution you'd like
+A clear and concise description of what you want to happen.
+
+### Describe alternatives you've considered
+A clear and concise description of any alternative solutions or features you've considered.
+
+### Additional context
+Add any other context or screenshots about the feature request here.
@@ -292,3 +292,5 @@ __pycache__/
 *.map

 /Tests/MQTTnet.TestApp.NetCore/RetainedMessages.json
+
+Build/NuGet/
@@ -11,8 +11,8 @@
    <requireLicenseAcceptance>false</requireLicenseAcceptance>
    <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</description>
    <releaseNotes>
-* [Core] Fixed wrong versions in nuget packages.
-* [Server] The TCP address is now reused when starting which should prevent "port in used" error when restarting.
+* [ManagedClient] Added support for persisted sessions (thansk to @PMExtra).
+* [Server] Added support for assigned client IDs (MQTTv5 only) (thanks to @bcrosnier).
    </releaseNotes>
    <copyright>Copyright Christian Kratky 2016-2019</copyright>
    <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags> 
@@ -56,4 +56,4 @@

    <!-- Everything above .NET Framework 4.6.1 will use the binaries from 4.6.1. -->
  </files>
-</package>
+</package>
@@ -12,6 +12,13 @@ Write-Host "Nuget version    = $nugetVersion"
 Write-Host "MSBuild path     = $msbuild"
 Write-Host

+# Build and execute tests
+&$msbuild ..\Tests\MQTTnet.Core.Tests\MQTTnet.Tests.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netcoreapp2.1" /verbosity:m
+&$msbuild ..\Tests\MQTTnet.AspNetCore.Tests\MQTTnet.AspNetCore.Tests.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netcoreapp2.1" /verbosity:m
+
+vstest.console.exe ..\Tests\MQTTnet.Core.Tests\bin\Release\netcoreapp2.1\MQTTnet.Tests.dll
+vstest.console.exe ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp2.1\MQTTnet.AspNetCore.Tests.dll
+
 # Build the core library
 &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
@@ -44,7 +51,6 @@ Write-Host
 &$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"

 # Create NuGet packages.
-
 Invoke-WebRequest -Uri "https://dist.nuget.org/win-x86-commandline/latest/nuget.exe" -OutFile "nuget.exe"

 Remove-Item .\NuGet -Force -Recurse -ErrorAction SilentlyContinue
@@ -59,12 +65,12 @@ Copy-Item MQTTnet.Extensions.WebSocket4Net.nuspec -Destination MQTTnet.Extension
 (Get-Content MQTTnet.Extensions.WebSocket4Net.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.Extensions.WebSocket4Net.nuspec

 New-Item -ItemType Directory -Force -Path .\NuGet
-.\nuget.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-.\nuget.exe pack MQTTnet.NETStandard.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-.\nuget.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-.\nuget.exe pack MQTTnet.Extensions.Rpc.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-.\nuget.exe pack MQTTnet.Extensions.ManagedClient.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-.\nuget.exe pack MQTTnet.Extensions.WebSocket4Net.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.NETStandard.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.Extensions.Rpc.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.Extensions.ManagedClient.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.Extensions.WebSocket4Net.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion

 Move-Item MQTTnet.AspNetCore.nuspec.old -Destination MQTTnet.AspNetCore.nuspec -Force
 Move-Item MQTTnet.Extensions.Rpc.nuspec.old -Destination MQTTnet.Extensions.Rpc.nuspec -Force
@@ -7,7 +7,7 @@ foreach ($file in $files)
 {
 	Write-Host "Uploading: " $file

-	.\nuget.exe push $file.Fullname $apiKey -NoSymbols -Source https://api.nuget.org/v3/index.json
+	.\nuget.exe push $file.Fullname $apiKey -Source https://api.nuget.org/v3/index.json
 }

 Remove-Item "nuget.exe" -Force -Recurse -ErrorAction SilentlyContinue
@@ -0,0 +1,239 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<!-- Created with Inkscape (http://www.inkscape.org/) -->
+
+<svg
+   xmlns:dc="http://purl.org/dc/elements/1.1/"
+   xmlns:cc="http://creativecommons.org/ns#"
+   xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+   xmlns:svg="http://www.w3.org/2000/svg"
+   xmlns="http://www.w3.org/2000/svg"
+   xmlns:sodipodi="http://sodipodi.sourceforge.net/DTD/sodipodi-0.dtd"
+   xmlns:inkscape="http://www.inkscape.org/namespaces/inkscape"
+   width="210mm"
+   height="297mm"
+   viewBox="0 0 210 297"
+   version="1.1"
+   id="svg8"
+   inkscape:version="0.92.4 (5da689c313, 2019-01-14)"
+   sodipodi:docname="MQTT.svg">
+  <defs
+     id="defs2">
+    <filter
+       inkscape:label="Blur Double"
+       inkscape:menu="Blurs"
+       inkscape:menu-tooltip="Overlays two copies with different blur amounts and modifiable blend and composite"
+       style="color-interpolation-filters:sRGB;"
+       id="filter3801">
+      <feGaussianBlur
+         stdDeviation="5"
+         result="fbSourceGraphic"
+         id="feGaussianBlur3793" />
+      <feGaussianBlur
+         stdDeviation="0.01"
+         in="SourceGraphic"
+         result="result1"
+         id="feGaussianBlur3795" />
+      <feComposite
+         in2="result1"
+         operator="arithmetic"
+         in="fbSourceGraphic"
+         k2="0.5"
+         k3="0.5"
+         result="result2"
+         id="feComposite3797" />
+      <feBlend
+         in2="fbSourceGraphic"
+         mode="normal"
+         result="result3"
+         id="feBlend3799" />
+    </filter>
+  </defs>
+  <sodipodi:namedview
+     id="base"
+     pagecolor="#ffffff"
+     bordercolor="#666666"
+     borderopacity="1.0"
+     inkscape:pageopacity="0.0"
+     inkscape:pageshadow="2"
+     inkscape:zoom="0.24748737"
+     inkscape:cx="955.72999"
+     inkscape:cy="295.38958"
+     inkscape:document-units="mm"
+     inkscape:current-layer="layer1"
+     showgrid="false"
+     inkscape:window-width="3840"
+     inkscape:window-height="2066"
+     inkscape:window-x="2989"
+     inkscape:window-y="-11"
+     inkscape:window-maximized="1" />
+  <metadata
+     id="metadata5">
+    <rdf:RDF>
+      <cc:Work
+         rdf:about="">
+        <dc:format>image/svg+xml</dc:format>
+        <dc:type
+           rdf:resource="http://purl.org/dc/dcmitype/StillImage" />
+        <dc:title />
+      </cc:Work>
+    </rdf:RDF>
+  </metadata>
+  <g
+     inkscape:label="Layer 1"
+     inkscape:groupmode="layer"
+     id="layer1">
+    <rect
+       ry="50"
+       rx="50"
+       style="fill:#000000;stroke-width:0.26458332;filter:url(#filter3801)"
+       y="92.501205"
+       x="56.367012"
+       height="512"
+       width="512"
+       id="rect3791"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001" />
+    <rect
+       id="rect3713"
+       width="512"
+       height="512"
+       x="53.721176"
+       y="88.797035"
+       style="fill:#aa0000;stroke-width:0.26458332"
+       rx="50"
+       ry="50"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001" />
+    <rect
+       style="fill:#ce0000;fill-opacity:0.60591131;stroke-width:0.28100395"
+       id="rect3806"
+       width="143.25298"
+       height="9.0714283"
+       x="270.81995"
+       y="554.02386"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001" />
+    <path
+       inkscape:connector-curvature="0"
+       id="path4638"
+       d="m 494.14104,142.00092 -5.35736,41.28822 67.77684,374.38977 a 263.46279,229.17651 36.835702 0 1 4.98278,-0.4605 263.46279,229.17651 36.835702 0 1 7.77403,0.0471 z"
+       style="fill:#ce0000;fill-opacity:0.60591131;stroke-width:0.59952348"
+       inkscape:transform-center-x="-28.818481"
+       inkscape:transform-center-y="146.97441"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001" />
+    <path
+       style="fill:#d40000;stroke-width:1.76103246"
+       d="m 2138.1582,1411.377 a 995.76488,866.17897 0 0 0 -842.3984,854.3418 995.76488,866.17897 0 0 0 0.2988,5.0097 h 653.123 c 104.693,0 188.9766,-84.2836 188.9766,-188.9765 z"
+       transform="scale(0.26458333)"
+       id="path3715-5-8"
+       inkscape:connector-curvature="0"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001" />
+    <ellipse
+       style="fill:#d40000;stroke-width:0.06901014"
+       id="path3715-5"
+       cx="492.58557"
+       cy="170.58017"
+       rx="39.021355"
+       ry="33.94323"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001" />
+    <path
+       style="fill:#ce0000;fill-opacity:0.60591131;stroke-width:2.26591563"
+       d="M 464.66602,532.89453 542.01562,669.92969 1595.373,1648.873 a 995.76487,866.17896 0 0 1 14.0293,-12.6835 995.76487,866.17896 0 0 1 23.6231,-17.4727 z"
+       id="rect3806-2"
+       inkscape:connector-curvature="0"
+       transform="scale(0.26458333)"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001" />
+    <text
+       xml:space="preserve"
+       style="font-style:normal;font-weight:normal;font-size:10.58333302px;line-height:1.25;font-family:sans-serif;letter-spacing:0px;word-spacing:0px;fill:#000000;fill-opacity:1;stroke:none;stroke-width:0.26458332"
+       x="102.34985"
+       y="320.75201"
+       id="text3774"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001"><tspan
+         sodipodi:role="line"
+         id="tspan3772"
+         x="102.34985"
+         y="320.75201"
+         style="font-size:141.1111145px;fill:#ffffff;stroke-width:0.26458332">MQTT</tspan></text>
+    <ellipse
+       ry="33.94323"
+       rx="39.021355"
+       cy="148.56271"
+       cx="124.47787"
+       id="ellipse3768"
+       style="fill:#d40000;stroke-width:0.06901014"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001" />
+    <ellipse
+       style="fill:#d40000;stroke-width:0.06901014"
+       id="ellipse3770"
+       cx="234.9953"
+       cy="556.91571"
+       rx="39.021355"
+       ry="33.94323"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001" />
+    <text
+       id="text3778"
+       y="474.73837"
+       x="194.57605"
+       style="font-style:normal;font-weight:normal;font-size:10.58333302px;line-height:1.25;font-family:sans-serif;letter-spacing:0px;word-spacing:0px;fill:#000000;fill-opacity:1;stroke:none;stroke-width:0.26458332"
+       xml:space="preserve"
+       inkscape:export-xdpi="2.6500001"
+       inkscape:export-ydpi="2.6500001"><tspan
+         style="font-size:141.1111145px;fill:#ffffff;stroke-width:0.26458332"
+         y="474.73837"
+         x="194.57605"
+         id="tspan3776"
+         sodipodi:role="line">net</tspan></text>
+    <rect
+       ry="50"
+       rx="50"
+       style="fill:#aa0000;stroke-width:0.26458332"
+       y="97.349655"
+       x="-672.18286"
+       height="512"
+       width="512"
+       id="rect4642"
+       inkscape:export-xdpi="6.3499999"
+       inkscape:export-ydpi="6.3499999" />
+    <path
+       inkscape:connector-curvature="0"
+       id="path4648"
+       d="m -160.18302,381.97945 a 263.46279,229.17652 0 0 0 -222.88457,226.0446 263.46279,229.17652 0 0 0 0.0791,1.32549 h 172.80546 c 27.70002,0 50.00005,-22.30004 50.00005,-50.00004 z"
+       style="fill:#d40000;stroke-width:0.46593982"
+       inkscape:export-xdpi="6.3499999"
+       inkscape:export-ydpi="6.3499999" />
+    <text
+       id="text4656"
+       y="329.30466"
+       x="-639.42944"
+       style="font-style:normal;font-weight:normal;font-size:10.58333302px;line-height:1.25;font-family:sans-serif;letter-spacing:0px;word-spacing:0px;fill:#000000;fill-opacity:1;stroke:none;stroke-width:0.26458332"
+       xml:space="preserve"
+       inkscape:export-xdpi="6.3499999"
+       inkscape:export-ydpi="6.3499999"><tspan
+         style="font-weight:bold;font-size:141.1111145px;fill:#ffffff;stroke-width:0.26458332"
+         y="329.30466"
+         x="-639.42944"
+         id="tspan4654"
+         sodipodi:role="line">MQTT</tspan></text>
+    <text
+       xml:space="preserve"
+       style="font-style:normal;font-weight:normal;font-size:10.58333302px;line-height:1.25;font-family:sans-serif;letter-spacing:0px;word-spacing:0px;fill:#000000;fill-opacity:1;stroke:none;stroke-width:0.26458332"
+       x="-550.90747"
+       y="482.23267"
+       id="text4664"
+       inkscape:export-xdpi="6.3499999"
+       inkscape:export-ydpi="6.3499999"><tspan
+         sodipodi:role="line"
+         id="tspan4662"
+         x="-550.90747"
+         y="482.23267"
+         style="font-weight:bold;font-size:141.1111145px;fill:#ffffff;stroke-width:0.26458332">net</tspan></text>
+  </g>
+</svg>
@@ -1,5 +1,5 @@
 <p align="center">
-<img src="https://github.com/chkr1011/MQTTnet/blob/master/Images/Logo_128x128.png?raw=true" width="128">
+<img src="https://github.com/chkr1011/MQTTnet/blob/master/Images/icon_det_256.png?raw=true" width="196">
 <br/>
 <br/>
 </p>
@@ -162,11 +162,12 @@ namespace MQTTnet.AspNetCore
                var buffer = formatter.Encode(packet);
                var msg = buffer.AsMemory();
                var output = _output;
-                msg.CopyTo(output.GetMemory(msg.Length));
-                BytesSent += msg.Length;
+                var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false);
+                if (result.IsCompleted)
+                {
+                    BytesSent += msg.Length;
+                }
                PacketFormatterAdapter.FreeBuffer();
-                output.Advance(msg.Length);
-                await output.FlushAsync().ConfigureAwait(false);
            }
            finally
            {
@@ -64,7 +64,6 @@ namespace MQTTnet.AspNetCore

            GrowIfNeeded(propertyWriter.Length);
            Write(propertyWriter.GetBuffer(), 0, propertyWriter.Length);
-            Commit(propertyWriter.Length);
        }

        public void Write(byte[] payload, int start, int length)
@@ -24,6 +24,8 @@ namespace MQTTnet.Extensions.ManagedClient

        private readonly IMqttClient _mqttClient;
        private readonly IMqttNetChildLogger _logger;
+        
+        private readonly AsyncLock _messageQueueLock = new AsyncLock();

        private CancellationTokenSource _connectionCancellationToken;
        private CancellationTokenSource _publishingCancellationToken;
@@ -147,7 +149,7 @@ namespace MQTTnet.Extensions.ManagedClient

            try
            {
-                lock (_messageQueue)
+                using (await _messageQueueLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
                {
                    if (_messageQueue.Count >= Options.MaxPendingMessages)
                    {
@@ -167,6 +169,16 @@ namespace MQTTnet.Extensions.ManagedClient
                    }

                    _messageQueue.Enqueue(applicationMessage);
+                    
+                    if (_storageManager != null)
+                    {
+                        if (removedMessage != null)
+                        {
+                            await _storageManager.RemoveAsync(removedMessage).ConfigureAwait(false);
+                        }
+
+                        await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
+                    }
                }
            }
            finally
@@ -181,16 +193,6 @@ namespace MQTTnet.Extensions.ManagedClient
                }

            }
-            
-            if (_storageManager != null)
-            {
-                if (removedMessage != null)
-                {
-                    await _storageManager.RemoveAsync(removedMessage).ConfigureAwait(false);
-                }
-
-                await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
-            }
        }

        public Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
@@ -250,6 +252,7 @@ namespace MQTTnet.Extensions.ManagedClient
                _maintainConnectionTask = null;
            }

+            _messageQueueLock.Dispose();
            _mqttClient.Dispose();
        }

@@ -362,7 +365,7 @@ namespace MQTTnet.Extensions.ManagedClient
            }
            catch (Exception exception)
            {
-                _logger.Error(exception, "Unhandled exception while publishing queued application messages.");
+                _logger.Error(exception, "Error while publishing queued application messages.");
            }
            finally
            {
@@ -377,7 +380,7 @@ namespace MQTTnet.Extensions.ManagedClient
            {
                await _mqttClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);

-                lock (_messageQueue) //lock to avoid conflict with this.PublishAsync
+                using (await _messageQueueLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync
                {
                    // While publishing this message, this.PublishAsync could have booted this
                    // message off the queue to make room for another (when using a cap
@@ -386,11 +389,11 @@ namespace MQTTnet.Extensions.ManagedClient
                    // it from the queue.  If not, that means this.PublishAsync has already
                    // removed it, in which case we don't want to do anything.
                    _messageQueue.RemoveFirst(i => i.Id.Equals(message.Id));
-                }
-
-                if (_storageManager != null)
-                {
-                    await _storageManager.RemoveAsync(message).ConfigureAwait(false);
+                    
+                    if (_storageManager != null)
+                    {
+                        await _storageManager.RemoveAsync(message).ConfigureAwait(false);
+                    }
                }
            }
            catch (MqttCommunicationException exception)
@@ -408,21 +411,21 @@ namespace MQTTnet.Extensions.ManagedClient
                    //contradict the expected behavior of QoS 1 and 2, that's also true
                    //for the usage of a message queue cap, so it's still consistent
                    //with prior behavior in that way.
-                    lock (_messageQueue) //lock to avoid conflict with this.PublishAsync
+                    using (await _messageQueueLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync
                    {
                        _messageQueue.RemoveFirst(i => i.Id.Equals(message.Id));
-                    }
-
-                    if (_storageManager != null)
-                    {
-                        await _storageManager.RemoveAsync(message).ConfigureAwait(false);
+                        
+                        if (_storageManager != null)
+                        {
+                            await _storageManager.RemoveAsync(message).ConfigureAwait(false);
+                        }
                    }
                }
            }
            catch (Exception exception)
            {
                transmitException = exception;
-                _logger.Error(exception, $"Unhandled exception while publishing application message ({message.Id}).");
+                _logger.Error(exception, $"Error while publishing application message ({message.Id}).");
            }
            finally
            {
@@ -533,4 +536,4 @@ namespace MQTTnet.Extensions.ManagedClient
            _connectionCancellationToken = null;
        }
    }
-}
+}
@@ -5,6 +5,8 @@ using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Client;
 using MQTTnet.Exceptions;
+using MQTTnet.Extensions.Rpc.Options;
+using MQTTnet.Extensions.Rpc.Options.TopicGeneration;
 using MQTTnet.Protocol;

 namespace MQTTnet.Extensions.Rpc
@@ -13,11 +15,18 @@ namespace MQTTnet.Extensions.Rpc
    {
        private readonly ConcurrentDictionary<string, TaskCompletionSource<byte[]>> _waitingCalls = new ConcurrentDictionary<string, TaskCompletionSource<byte[]>>();
        private readonly IMqttClient _mqttClient;
+        private readonly IMqttRpcClientOptions _options;
        private readonly RpcAwareApplicationMessageReceivedHandler _applicationMessageReceivedHandler;

-        public MqttRpcClient(IMqttClient mqttClient)
+        [Obsolete("Use MqttRpcClient(IMqttClient mqttClient, IMqttRpcClientOptions options).")]
+        public MqttRpcClient(IMqttClient mqttClient) : this(mqttClient, new MqttRpcClientOptions())
+        {
+        }
+
+        public MqttRpcClient(IMqttClient mqttClient, IMqttRpcClientOptions options)
        {
            _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
+            _options = options ?? throw new ArgumentNullException(nameof(options));

            _applicationMessageReceivedHandler = new RpcAwareApplicationMessageReceivedHandler(
                mqttClient.ApplicationMessageReceivedHandler,
@@ -55,8 +64,26 @@ namespace MQTTnet.Extensions.Rpc
                throw new InvalidOperationException("The application message received handler was modified.");
            }

-            var requestTopic = $"MQTTnet.RPC/{Guid.NewGuid():N}/{methodName}";
-            var responseTopic = requestTopic + "/response";
+            var topicNames = _options.TopicGenerationStrategy.CreateRpcTopics(new TopicGenerationContext
+            {
+                MethodName = methodName,
+                QualityOfServiceLevel = qualityOfServiceLevel,
+                MqttClient = _mqttClient,
+                Options = _options
+            });
+
+            var requestTopic = topicNames.RequestTopic;
+            var responseTopic = topicNames.ResponseTopic;
+
+            if (string.IsNullOrWhiteSpace(requestTopic))
+            {
+                throw new MqttProtocolViolationException("RPC request topic is empty.");
+            }
+
+            if (string.IsNullOrWhiteSpace(responseTopic))
+            {
+                throw new MqttProtocolViolationException("RPC response topic is empty.");
+            }

            var requestMessage = new MqttApplicationMessageBuilder()
                .WithTopic(requestTopic)
@@ -0,0 +1,9 @@
+using MQTTnet.Extensions.Rpc.Options.TopicGeneration;
+
+namespace MQTTnet.Extensions.Rpc.Options
+{
+    public interface IMqttRpcClientOptions
+    {
+        IMqttRpcClientTopicGenerationStrategy TopicGenerationStrategy { get; set; }
+    }
+}
@@ -0,0 +1,9 @@
+using MQTTnet.Extensions.Rpc.Options.TopicGeneration;
+
+namespace MQTTnet.Extensions.Rpc.Options
+{
+    public class MqttRpcClientOptions : IMqttRpcClientOptions
+    {
+        public IMqttRpcClientTopicGenerationStrategy TopicGenerationStrategy { get; set; } = new DefaultMqttRpcClientTopicGenerationStrategy();
+    }
+}
@@ -0,0 +1,25 @@
+using MQTTnet.Extensions.Rpc.Options.TopicGeneration;
+using System;
+
+namespace MQTTnet.Extensions.Rpc.Options
+{
+    public class MqttRpcClientOptionsBuilder
+    {
+        IMqttRpcClientTopicGenerationStrategy _topicGenerationStrategy = new DefaultMqttRpcClientTopicGenerationStrategy();
+
+        public MqttRpcClientOptionsBuilder WithTopicGenerationStrategy(IMqttRpcClientTopicGenerationStrategy value)
+        {
+            _topicGenerationStrategy = value ?? throw new ArgumentNullException(nameof(value));
+
+            return this;
+        }
+
+        public IMqttRpcClientOptions Build()
+        {
+            return new MqttRpcClientOptions
+            {
+                TopicGenerationStrategy = _topicGenerationStrategy
+            };
+        }
+    }
+}
@@ -0,0 +1,9 @@
+namespace MQTTnet.Extensions.Rpc.Options
+{
+    public class MqttRpcTopicPair
+    {
+        public string RequestTopic { get; set; }
+
+        public string ResponseTopic { get; set; }
+    }
+}
@@ -0,0 +1,20 @@
+using MQTTnet.Extensions.Rpc.Options.TopicGeneration;
+using System;
+
+namespace MQTTnet.Extensions.Rpc.Options
+{
+    public class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy
+    {
+        public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
+        {
+            var requestTopic = $"MQTTnet.RPC/{Guid.NewGuid():N}/{context.MethodName}";
+            var responseTopic = requestTopic + "/response";
+
+            return new MqttRpcTopicPair
+            {
+                RequestTopic = requestTopic,
+                ResponseTopic = responseTopic
+            };
+        }
+    }
+}
@@ -0,0 +1,7 @@
+namespace MQTTnet.Extensions.Rpc.Options.TopicGeneration
+{
+    public interface IMqttRpcClientTopicGenerationStrategy
+    {
+        MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context);
+    }
+}
@@ -0,0 +1,16 @@
+using MQTTnet.Client;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Extensions.Rpc.Options.TopicGeneration
+{
+    public class TopicGenerationContext
+    {
+        public string MethodName { get; set; }
+
+        public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
+
+        public IMqttClient MqttClient { get; set; }
+
+        public IMqttRpcClientOptions Options { get; set; }
+    }
+}
@@ -0,0 +1,35 @@
+using System.IO;
+
+namespace MQTTnet.Server.Configuration
+{
+    public class CertificateSettingsModel
+    {
+        /// <summary>
+        /// Path to certificate.
+        /// </summary>
+        public string Path { get; set; }
+        
+        /// <summary>
+        /// Password of certificate.
+        /// </summary>
+        public string Password { get; set; }
+
+        /// <summary>
+        /// Read certificate file.
+        /// </summary>
+        public byte[] ReadCertificate()
+        {
+            if (string.IsNullOrEmpty(Path) || string.IsNullOrWhiteSpace(Path))
+            {
+                throw new FileNotFoundException("No path set");
+            }
+
+            if (!File.Exists(Path))
+            {
+                throw new FileNotFoundException($"Could not find Certificate in path: {Path}");
+            }
+
+            return File.ReadAllBytes(Path);
+        }
+    }
+}
@@ -9,9 +9,9 @@ namespace MQTTnet.Server.Configuration
    public class TcpEndPointModel
    {
        /// <summary>
-        /// Path to Certificate
+        /// Certificate settings.
        /// </summary>
-        public string CertificatePath { get; set; }
+        public CertificateSettingsModel Certificate { get; set; }

        /// <summary>
        /// Enabled / Disable
@@ -33,25 +33,6 @@ namespace MQTTnet.Server.Configuration
        /// </summary>
        public int Port { get; set; } = 1883;

-        /// <summary>
-        /// Read Certificate file
-        /// </summary>
-        /// <returns></returns>
-        public byte[] ReadCertificate()
-        {
-            if (string.IsNullOrEmpty(CertificatePath) || string.IsNullOrWhiteSpace(CertificatePath))
-            {
-                throw new FileNotFoundException("No path set");
-            }
-
-            if (!File.Exists(CertificatePath))
-            {
-                throw new FileNotFoundException($"Could not find Certificate in path: {CertificatePath}");
-            }
-
-            return File.ReadAllBytes(CertificatePath);
-        }
-
        /// <summary>
        /// Read IPv4
        /// </summary>
@@ -13,7 +13,7 @@ namespace MQTTnet.Server.Controllers
 {
    [Authorize]
    [ApiController]
-    public class ClientsController : ControllerBase
+    public class ClientsController : Controller
    {
        private readonly MqttServerService _mqttServerService;

@@ -0,0 +1,51 @@
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Mvc;
+using MQTTnet.Protocol;
+using MQTTnet.Server.Mqtt;
+
+namespace MQTTnet.Server.Controllers
+{
+    [Authorize]
+    [ApiController]
+    public class MessagesController : Controller
+    {
+        private readonly MqttServerService _mqttServerService;
+
+        public MessagesController(MqttServerService mqttServerService)
+        {
+            _mqttServerService = mqttServerService ?? throw new ArgumentNullException(nameof(mqttServerService));
+        }
+
+        [Route("api/v1/messages")]
+        [HttpPost]
+        public async Task<ActionResult> PostMessage(MqttApplicationMessage message)
+        {
+            await _mqttServerService.PublishAsync(message);
+            return Ok();
+        }
+
+        [Route("api/v1/messages/{*topic}")]
+        [HttpPost]
+        public async Task<ActionResult> PostMessage(string topic, int qosLevel = 0)
+        {
+            byte[] payload;
+
+            using (var memoryStream = new MemoryStream())
+            {
+                await HttpContext.Request.Body.CopyToAsync(memoryStream);
+                payload = memoryStream.ToArray();
+            }
+
+            var message = new MqttApplicationMessageBuilder()
+                .WithTopic(topic)
+                .WithPayload(payload)
+                .WithQualityOfServiceLevel((MqttQualityOfServiceLevel)qosLevel)
+                .Build();
+
+            return await PostMessage(message);
+        }
+    }
+}
@@ -12,7 +12,7 @@ namespace MQTTnet.Server.Controllers
 {
    [Authorize]
    [ApiController]
-    public class RetainedApplicationMessagesController : ControllerBase
+    public class RetainedApplicationMessagesController : Controller
    {
        private readonly MqttServerService _mqttServerService;

@@ -0,0 +1,18 @@
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Mvc;
+using System.Reflection;
+
+namespace MQTTnet.Server.Controllers
+{
+    [Authorize]
+    [ApiController]
+    public class ServerController : Controller
+    {
+        [Route("api/v1/server/version")]
+        [HttpGet]
+        public ActionResult<string> GetVersion()
+        {
+            return Assembly.GetExecutingAssembly().GetCustomAttribute<AssemblyInformationalVersionAttribute>().InformationalVersion;
+        }
+    }
+}
@@ -13,7 +13,7 @@ namespace MQTTnet.Server.Controllers
 {
    [Authorize]
    [ApiController]
-    public class SessionsController : ControllerBase
+    public class SessionsController : Controller
    {
        private readonly MqttServerService _mqttServerService;

@@ -22,14 +22,18 @@ namespace MQTTnet.Server.Mqtt
        {
            try
            {
+                // This might be not set when a message was published by the server instead of a client.
+                context.SessionItems.TryGetValue(MqttServerConnectionValidator.WrappedSessionItemsKey, out var sessionItems);
+
                var pythonContext = new PythonDictionary
                {
+                    { "client_id", context.ClientId },
+                    { "session_items", sessionItems },
+                    { "retain", context.ApplicationMessage.Retain },
                    { "accept_publish", context.AcceptPublish },
                    { "close_connection", context.CloseConnection },
-                    { "client_id", context.ClientId },
                    { "topic", context.ApplicationMessage.Topic },
-                    { "qos", (int)context.ApplicationMessage.QualityOfServiceLevel },
-                    { "retain", context.ApplicationMessage.Retain }
+                    { "qos", (int)context.ApplicationMessage.QualityOfServiceLevel }
                };
                
                _pythonScriptHostService.InvokeOptionalFunction("on_intercept_application_message", pythonContext);
@@ -47,4 +51,4 @@ namespace MQTTnet.Server.Mqtt
            return Task.CompletedTask;
        }
    }
-}
+}
@@ -9,6 +9,8 @@ namespace MQTTnet.Server.Mqtt
 {
    public class MqttServerConnectionValidator : IMqttServerConnectionValidator
    {
+        public const string WrappedSessionItemsKey = "WRAPPED_ITEMS";
+
        private readonly PythonScriptHostService _pythonScriptHostService;
        private readonly ILogger _logger;

@@ -22,6 +24,8 @@ namespace MQTTnet.Server.Mqtt
        {
            try
            {
+                var sessionItems = new PythonDictionary();
+
                var pythonContext = new PythonDictionary
                {
                    { "endpoint", context.Endpoint },
@@ -33,6 +37,7 @@ namespace MQTTnet.Server.Mqtt
                    { "clean_session", context.CleanSession},
                    { "authentication_method", context.AuthenticationMethod},
                    { "authentication_data", new Bytes(context.AuthenticationData ?? new byte[0]) },
+                    { "session_items", sessionItems },

                    { "result", PythonConvert.Pythonfy(context.ReasonCode) }
                };
@@ -40,6 +45,8 @@ namespace MQTTnet.Server.Mqtt
                _pythonScriptHostService.InvokeOptionalFunction("on_validate_client_connection", pythonContext);

                context.ReasonCode = PythonConvert.ParseEnum<MqttConnectReasonCode>((string)pythonContext["result"]);
+
+                context.SessionItems[WrappedSessionItemsKey] = sessionItems;
            }
            catch (Exception exception)
            {
@@ -47,7 +47,7 @@ namespace MQTTnet.Server.Mqtt
            MqttSubscriptionInterceptor mqttSubscriptionInterceptor,
            MqttApplicationMessageInterceptor mqttApplicationMessageInterceptor,
            MqttServerStorage mqttServerStorage,
-            PythonScriptHostService pythonScriptHostService,            
+            PythonScriptHostService pythonScriptHostService,
            ILogger<MqttServerService> logger)
        {
            _settings = mqttSettings ?? throw new ArgumentNullException(nameof(mqttSettings));
@@ -179,7 +179,7 @@ namespace MQTTnet.Server.Mqtt
                .WithApplicationMessageInterceptor(_mqttApplicationMessageInterceptor)
                .WithSubscriptionInterceptor(_mqttSubscriptionInterceptor)
                .WithStorage(_mqttServerStorage);
-            
+
            // Configure unencrypted connections
            if (_settings.TcpEndPoint.Enabled)
            {
@@ -210,9 +210,23 @@ namespace MQTTnet.Server.Mqtt
            {
                options
                    .WithEncryptedEndpoint()
-                    .WithEncryptionSslProtocol(SslProtocols.Tls12)
-                    .WithEncryptionCertificate(_settings.EncryptedTcpEndPoint.ReadCertificate());
+                    .WithEncryptionSslProtocol(SslProtocols.Tls12);
+                
+                if (!string.IsNullOrEmpty(_settings.EncryptedTcpEndPoint?.Certificate?.Path))
+                {
+                    IMqttServerCertificateCredentials certificateCredentials = null;

+                    if (!string.IsNullOrEmpty(_settings.EncryptedTcpEndPoint?.Certificate?.Password))
+                    {
+                        certificateCredentials = new MqttServerCertificateCredentials
+                        {
+                            Password = _settings.EncryptedTcpEndPoint.Certificate.Password
+                        };
+                    }
+
+                    options.WithEncryptionCertificate(_settings.EncryptedTcpEndPoint.Certificate.ReadCertificate(), certificateCredentials);
+                }
+                
                if (_settings.EncryptedTcpEndPoint.TryReadIPv4(out var address4))
                {
                    options.WithEncryptedEndpointBoundIPAddress(address4);
@@ -21,14 +21,16 @@ namespace MQTTnet.Server.Mqtt
        {
            try
            {
+                var sessionItems = (PythonDictionary)context.SessionItems[MqttServerConnectionValidator.WrappedSessionItemsKey];
+
                var pythonContext = new PythonDictionary
                {
-                    { "accept_subscription", context.AcceptSubscription },
-                    { "close_connection", context.CloseConnection },
-
                    { "client_id", context.ClientId },
+                    { "session_items", sessionItems },
                    { "topic", context.TopicFilter.Topic },
-                    { "qos", (int)context.TopicFilter.QualityOfServiceLevel }
+                    { "qos", (int)context.TopicFilter.QualityOfServiceLevel },
+                    { "accept_subscription", context.AcceptSubscription },
+                    { "close_connection", context.CloseConnection }
                };

                _pythonScriptHostService.InvokeOptionalFunction("on_intercept_subscription", pythonContext);
@@ -27,7 +27,10 @@
            "IPv4": "*",
            "IPv6": "*",
            "Port": 8883,
-            "CertificatePath": "/absolute/path/to/pfx"
+            "Certificate": {
+                "Path": "/absolute/path/to/pfx",
+                "Password": ""
+            }
        },
        "WebSocketEndPoint": {
            "Enabled": true,
@@ -63,4 +66,4 @@
        }
    },
    "AllowedHosts": "*"
-}
+}
@@ -257,12 +257,12 @@ namespace MQTTnet.Client
        {
            var clientWasConnected = IsConnected;

-            InitiateDisconnect();
-
-            IsConnected = false;
+            TryInitiateDisconnect();

            try
            {
+                IsConnected = false;
+
                if (_adapter != null)
                {
                    _logger.Verbose("Disconnecting [Timeout={0}]", Options.CommunicationTimeout);
@@ -295,7 +295,7 @@ namespace MQTTnet.Client
            }
        }

-        private void InitiateDisconnect()
+        private void TryInitiateDisconnect()
        {
            lock (_disconnectLock)
            {
@@ -139,6 +139,13 @@ namespace MQTTnet.Client.Options
            return this;
        }

+        public MqttClientOptionsBuilder WithCredentials(IMqttClientCredentials credentials)
+        {
+            _options.Credentials = credentials;
+
+            return this;
+        }
+
        public MqttClientOptionsBuilder WithExtendedAuthenticationExchangeHandler(IMqttExtendedAuthenticationExchangeHandler handler)
        {
            _options.ExtendedAuthenticationExchangeHandler = handler;
@@ -5,15 +5,25 @@
        int Length { get; }

        void WriteWithLengthPrefix(string value);
-        void Write(byte returnCode);
-        void WriteWithLengthPrefix(byte[] payload);
-        void Write(ushort keepAlivePeriod);
-        void Write(IMqttPacketWriter propertyWriter);
-        void WriteVariableLengthInteger(uint length);
-        void Write(byte[] payload, int v, int length);
-        void Reset(int v);
-        void Seek(int v);
+
+        void Write(byte value);
+
+        void WriteWithLengthPrefix(byte[] value);
+
+        void Write(ushort value);
+
+        void Write(IMqttPacketWriter value);
+
+        void WriteVariableLengthInteger(uint value);
+
+        void Write(byte[] value, int offset, int length);
+
+        void Reset(int length);
+
+        void Seek(int offset);
+
        void FreeBuffer();
+
        byte[] GetBuffer();
    }
 }
@@ -21,8 +21,6 @@ namespace MQTTnet.Formatter
        {
            // The MQTT fixed header contains 1 byte of flags and at least 1 byte for the remaining data length.
            // So in all cases at least 2 bytes must be read for a complete MQTT packet.
-            // async/await is used here because the next packet is received in a couple of minutes so the performance
-            // impact is acceptable according to a useless waiting thread.
            var buffer = fixedHeaderBuffer;
            var totalBytesRead = 0;

@@ -55,16 +53,7 @@ namespace MQTTnet.Formatter
                };
            }

-#if WINDOWS_UWP
-            // UWP will have a dead lock when calling this not async.
            var bodyLength = await ReadBodyLengthAsync(buffer[1], cancellationToken).ConfigureAwait(false);
-#else
-            // Here the async/await pattern is not used because the overhead of context switches
-            // is too big for reading 1 byte in a row. We expect that the remaining data was sent
-            // directly after the initial bytes. If the client disconnects just in this moment we
-            // will get an exception anyway.
-            var bodyLength = ReadBodyLength(buffer[1], cancellationToken);
-#endif

            if (!bodyLength.HasValue)
            {
@@ -81,49 +70,6 @@ namespace MQTTnet.Formatter
            };
        }

-#if !WINDOWS_UWP
-        private int? ReadBodyLength(byte initialEncodedByte, CancellationToken cancellationToken)
-        {
-            var offset = 0;
-            var multiplier = 128;
-            var value = initialEncodedByte & 127;
-            int encodedByte = initialEncodedByte;
-
-            while ((encodedByte & 128) != 0)
-            {
-                offset++;
-                if (offset > 3)
-                {
-                    throw new MqttProtocolViolationException("Remaining length is invalid.");
-                }
-
-                if (cancellationToken.IsCancellationRequested)
-                {
-                    return null;
-                }
-
-                var readCount = _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).GetAwaiter().GetResult();
-
-                if (cancellationToken.IsCancellationRequested)
-                {
-                    return null;
-                }
-
-                if (readCount == 0)
-                {
-                    return null;
-                }
-
-                encodedByte = _singleByteBuffer[0];
-
-                value += (encodedByte & 127) * multiplier;
-                multiplier *= 128;
-            }
-
-            return value;
-        }
-#else
-
        private async Task<int?> ReadBodyLengthAsync(byte initialEncodedByte, CancellationToken cancellationToken)
        {
            var offset = 0;
@@ -164,6 +110,5 @@ namespace MQTTnet.Formatter

            return value;
        }
-#endif
    }
 }
@@ -328,10 +328,7 @@ namespace MQTTnet.Formatter.V5
            {
                ThrowReasonCodeNotSetException();
            }
-
-            packetWriter.Write(packet.PacketIdentifier.Value);
-            packetWriter.Write((byte)packet.ReasonCode.Value);
-
+           
            var propertiesWriter = new MqttV500PropertiesWriter();
            if (packet.Properties != null)
            {
@@ -339,6 +336,8 @@ namespace MQTTnet.Formatter.V5
                propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
            }

+            packetWriter.Write(packet.PacketIdentifier.Value);
+
            if (packetWriter.Length > 0 || packet.ReasonCode.Value != MqttPubRecReasonCode.Success)
            {
                packetWriter.Write((byte)packet.ReasonCode.Value);
@@ -84,9 +84,9 @@ namespace MQTTnet.Implementations
            if (_options.TlsOptions.UseTls)
            {
                var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
-                await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
-
                _stream = sslStream;
+
+                await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, !_options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);                
            }
            else
            {
@@ -41,14 +41,24 @@ namespace MQTTnet.Implementations
                RegisterListeners(options.DefaultEndpointOptions, null, _cancellationTokenSource.Token);
            }

-            if (options.TlsEndpointOptions.IsEnabled)
+            if (options.TlsEndpointOptions?.IsEnabled == true)
            {
                if (options.TlsEndpointOptions.Certificate == null)
                {
                    throw new ArgumentException("TLS certificate is not set.");
                }

-                var tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate);
+                X509Certificate2 tlsCertificate;
+                if (string.IsNullOrEmpty(options.TlsEndpointOptions.CertificateCredentials?.Password))
+                {
+                    // Use a different overload when no password is specified. Otherwise the constructor will fail.
+                    tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate);
+                }
+                else
+                {
+                    tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate, options.TlsEndpointOptions.CertificateCredentials.Password);
+                }
+                
                if (!tlsCertificate.HasPrivateKey)
                {
                    throw new InvalidOperationException("The certificate for TLS encryption must contain the private key.");
@@ -60,8 +60,19 @@ namespace MQTTnet.Implementations
                _logger.Info($"Starting TCP listener for {_localEndPoint} TLS={_tlsCertificate != null}.");

                _socket = new Socket(_addressFamily, SocketType.Stream, ProtocolType.Tcp);
-                _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
-                _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.NoDelay, true);
+
+                // Usage of socket options is described here: https://docs.microsoft.com/en-us/dotnet/api/system.net.sockets.socket.setsocketoption?view=netcore-2.2
+
+                if (_options.ReuseAddress)
+                {
+                    _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
+                }
+                
+                if (_options.NoDelay)
+                {
+                    _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, true);
+                }
+                
                _socket.Bind(_localEndPoint);
                _socket.Listen(_options.ConnectionBacklog);

@@ -151,7 +162,7 @@ namespace MQTTnet.Implementations

                if (_tlsCertificate != null)
                {
-                    var sslStream = new SslStream(stream, false);
+                    var sslStream = new SslStream(stream, false, _tlsOptions.RemoteCertificateValidationCallback);

                    await sslStream.AuthenticateAsServerAsync(
                        _tlsCertificate, 
@@ -162,6 +173,11 @@ namespace MQTTnet.Implementations
                    stream = sslStream;

                    clientCertificate = sslStream.RemoteCertificate as X509Certificate2;
+
+                    if (clientCertificate == null && sslStream.RemoteCertificate != null)
+                    {
+                        clientCertificate = new X509Certificate2(sslStream.RemoteCertificate.Export(X509ContentType.Cert));
+                    }
                }

                var clientHandler = ClientHandler;
@@ -23,7 +23,7 @@ namespace MQTTnet.Internal
        public Task<IDisposable> WaitAsync(CancellationToken cancellationToken)
        {
            var task = _semaphore.WaitAsync(cancellationToken);
-            if (task.IsCompleted)
+            if (task.Status == TaskStatus.RanToCompletion)
            {
                return _releaser;
            }
@@ -41,13 +41,13 @@
    <DefineConstants>RELEASE;NETSTANDARD1_3</DefineConstants>
  </PropertyGroup>

-  <ItemGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
+  <ItemGroup Condition="'$(TargetFramework)'=='netstandard1.3'">
    <PackageReference Include="System.Net.Security" Version="4.3.2" />
    <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
    <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.2" />
  </ItemGroup>

-  <ItemGroup Condition="'$(TargetFramework)'=='netstandard1.3'">
+  <ItemGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
    <PackageReference Include="System.Net.Security" Version="4.3.2" />
    <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
    <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.2" />
@@ -13,5 +13,31 @@ namespace MQTTnet.Packets
        public string Name { get; }

        public string Value { get; }
+
+        public override int GetHashCode()
+        {
+            return Name.GetHashCode() ^ Value.GetHashCode();
+        }
+
+        public override bool Equals(object other)
+        {
+            return Equals(other as MqttUserProperty);
+        }
+
+        public bool Equals(MqttUserProperty other)
+        {
+            if (other == null)
+            {
+                return false;
+            }
+
+            if (ReferenceEquals(other, this))
+            {
+                return true;
+            }
+
+            return string.Equals(Name, other.Name, StringComparison.Ordinal) &&
+                string.Equals(Value, other.Value, StringComparison.Ordinal);
+        }
    }
 }
@@ -0,0 +1,4 @@
+public interface IMqttServerCertificateCredentials
+{
+    string Password { get; }
+}
@@ -21,6 +21,8 @@ namespace MQTTnet.Server
        MqttServerTcpEndpointOptions DefaultEndpointOptions { get; }
        MqttServerTlsTcpEndpointOptions TlsEndpointOptions { get; }

-        IMqttServerStorage Storage { get; }        
+        IMqttServerStorage Storage { get; }       
+        
+
    }
 }
@@ -0,0 +1,22 @@
+using System;
+using System.Collections.Generic;
+
+namespace MQTTnet.Server
+{
+    public interface IMqttServerPersistedSession
+    {
+        string ClientId { get; }
+
+        IDictionary<object, object> Items { get; }
+
+        IList<TopicFilter> Subscriptions { get; }
+
+        MqttApplicationMessage WillMessage { get; }
+
+        uint? WillDelayInterval { get; }
+
+        DateTime? SessionExpiryTimestamp { get; }
+
+        IList<MqttQueuedApplicationMessage> PendingApplicationMessages { get; }
+    }
+}
@@ -0,0 +1,11 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Server
+{
+    public interface IMqttServerPersistedSessionsStorage
+    {
+        Task<IList<IMqttServerPersistedSession>> LoadPersistedSessionsAsync();
+    }
+}
@@ -1,17 +1,25 @@
-namespace MQTTnet.Server
+using System.Collections.Generic;
+
+namespace MQTTnet.Server
 {
    public class MqttApplicationMessageInterceptorContext
    {
-        public MqttApplicationMessageInterceptorContext(string clientId, MqttApplicationMessage applicationMessage)
+        public MqttApplicationMessageInterceptorContext(string clientId, IDictionary<object, object> sessionItems, MqttApplicationMessage applicationMessage)
        {
            ClientId = clientId;
            ApplicationMessage = applicationMessage;
+            SessionItems = sessionItems;
        }

        public string ClientId { get; }

        public MqttApplicationMessage ApplicationMessage { get; set; }

+        /// <summary>
+        /// Gets or sets a key/value collection that can be used to share data within the scope of this session.
+        /// </summary>
+        public IDictionary<object, object> SessionItems { get; }
+
        public bool AcceptPublish { get; set; } = true;

        public bool CloseConnection { get; set; }
@@ -31,7 +31,6 @@ namespace MQTTnet.Server
        private readonly IMqttChannelAdapter _channelAdapter;
        private readonly IMqttDataConverter _dataConverter;
        private readonly string _endpoint;
-        private readonly MqttConnectPacket _connectPacket;
        private readonly DateTime _connectedTimestamp;

        private Task<MqttClientDisconnectType> _packageReceiverTask;
@@ -60,22 +59,24 @@ namespace MQTTnet.Server
            _channelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
            _dataConverter = _channelAdapter.PacketFormatterAdapter.DataConverter;
            _endpoint = _channelAdapter.Endpoint;
-            _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
+            ConnectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));

            if (logger == null) throw new ArgumentNullException(nameof(logger));
            _logger = logger.CreateChildLogger(nameof(MqttClientConnection));

-            _keepAliveMonitor = new MqttClientKeepAliveMonitor(_connectPacket.ClientId, StopAsync, _logger);
+            _keepAliveMonitor = new MqttClientKeepAliveMonitor(ConnectPacket.ClientId, StopAsync, _logger);

            _connectedTimestamp = DateTime.UtcNow;
            _lastPacketReceivedTimestamp = _connectedTimestamp;
            _lastNonKeepAlivePacketReceivedTimestamp = _lastPacketReceivedTimestamp;
        }

-        public string ClientId => _connectPacket.ClientId;
+        public MqttConnectPacket ConnectPacket { get; }
+
+        public string ClientId => ConnectPacket.ClientId;

        public MqttClientSession Session { get; }
-
+        
        public async Task StopAsync()
        {
            StopInternal();
@@ -133,12 +134,12 @@ namespace MQTTnet.Server
                _channelAdapter.ReadingPacketStartedCallback = OnAdapterReadingPacketStarted;
                _channelAdapter.ReadingPacketCompletedCallback = OnAdapterReadingPacketCompleted;

-                Session.WillMessage = _connectPacket.WillMessage;
+                Session.WillMessage = ConnectPacket.WillMessage;

                Task.Run(() => SendPendingPacketsAsync(_cancellationToken.Token), _cancellationToken.Token).Forget(_logger);

                // TODO: Change to single thread in SessionManager. Or use SessionManager and stats from KeepAliveMonitor.
-                _keepAliveMonitor.Start(_connectPacket.KeepAlivePeriod, _cancellationToken.Token);
+                _keepAliveMonitor.Start(ConnectPacket.KeepAlivePeriod, _cancellationToken.Token);

                await SendAsync(
                    new MqttConnAckPacket
@@ -228,7 +229,7 @@ namespace MQTTnet.Server
                }
                else
                {
-                    _logger.Error(exception, "Client '{0}': Unhandled exception while receiving client packets.", ClientId);
+                    _logger.Error(exception, "Client '{0}': Error while receiving client packets.", ClientId);
                }

                StopInternal();
@@ -271,7 +272,7 @@ namespace MQTTnet.Server
        private async Task HandleIncomingSubscribePacketAsync(MqttSubscribePacket subscribePacket)
        {
            // TODO: Let the channel adapter create the packet.
-            var subscribeResult = await Session.SubscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false);
+            var subscribeResult = await Session.SubscriptionsManager.SubscribeAsync(subscribePacket, ConnectPacket).ConfigureAwait(false);

            await SendAsync(subscribeResult.ResponsePacket).ConfigureAwait(false);

@@ -12,11 +12,12 @@ namespace MQTTnet.Server

        private readonly DateTime _createdTimestamp = DateTime.UtcNow;

-        public MqttClientSession(string clientId, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetChildLogger logger)
+        public MqttClientSession(string clientId, IDictionary<object, object> items, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetChildLogger logger)
        {
            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            Items = items ?? throw new ArgumentNullException(nameof(items));

-            SubscriptionsManager = new MqttClientSubscriptionsManager(clientId, eventDispatcher, serverOptions);
+            SubscriptionsManager = new MqttClientSubscriptionsManager(this, eventDispatcher, serverOptions);
            ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions);

            if (logger == null) throw new ArgumentNullException(nameof(logger));
@@ -33,6 +34,11 @@ namespace MQTTnet.Server

        public MqttClientSessionApplicationMessagesQueue ApplicationMessagesQueue { get; }

+        /// <summary>
+        /// Gets or sets a key/value collection that can be used to share data within the scope of this session.
+        /// </summary>
+        public IDictionary<object, object> Items { get; }
+
        public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage, string senderClientId, bool isRetainedApplicationMessage)
        {
            var checkSubscriptionsResult = SubscriptionsManager.CheckSubscriptions(applicationMessage.Topic, applicationMessage.QualityOfServiceLevel);
@@ -20,7 +20,8 @@ namespace MQTTnet.Server
        private readonly SemaphoreSlim _createConnectionGate = new SemaphoreSlim(1, 1);
        private readonly ConcurrentDictionary<string, MqttClientConnection> _connections = new ConcurrentDictionary<string, MqttClientConnection>();
        private readonly ConcurrentDictionary<string, MqttClientSession> _sessions = new ConcurrentDictionary<string, MqttClientSession>();
-        
+        private readonly IDictionary<object, object> _serverSessionItems = new ConcurrentDictionary<object, object>();
+
        private readonly CancellationToken _cancellationToken;
        private readonly MqttServerEventDispatcher _eventDispatcher;

@@ -241,19 +242,19 @@ namespace MQTTnet.Server

                clientId = connectPacket.ClientId;

-                var validatorContext = await ValidateConnectionAsync(connectPacket, channelAdapter).ConfigureAwait(false);
+                var connectionValidatorContext = await ValidateConnectionAsync(connectPacket, channelAdapter).ConfigureAwait(false);

-                if (validatorContext.ReasonCode != MqttConnectReasonCode.Success)
+                if (connectionValidatorContext.ReasonCode != MqttConnectReasonCode.Success)
                {
                    // Send failure response here without preparing a session. The result for a successful connect
                    // will be sent from the session itself.
-                    var connAckPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(validatorContext);
+                    var connAckPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext);
                    await channelAdapter.SendPacketAsync(connAckPacket, _options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false);

                    return;
                }

-                var connection = await CreateConnectionAsync(channelAdapter, connectPacket).ConfigureAwait(false);
+                var connection = await CreateConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false);

                await _eventDispatcher.HandleClientConnectedAsync(clientId).ConfigureAwait(false);
                
@@ -289,7 +290,7 @@ namespace MQTTnet.Server

        private async Task<MqttConnectionValidatorContext> ValidateConnectionAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
        {
-            var context = new MqttConnectionValidatorContext(connectPacket, channelAdapter);
+            var context = new MqttConnectionValidatorContext(connectPacket, channelAdapter, new ConcurrentDictionary<object, object>());

            var connectionValidator = _options.ConnectionValidator;

@@ -302,8 +303,7 @@ namespace MQTTnet.Server
            await connectionValidator.ValidateConnectionAsync(context).ConfigureAwait(false);

            // Check the client ID and set a random one if supported.
-            if (string.IsNullOrEmpty(connectPacket.ClientId) &&
-                channelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
+            if (string.IsNullOrEmpty(connectPacket.ClientId) && channelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
            {
                connectPacket.ClientId = context.AssignedClientIdentifier;
            }
@@ -316,7 +316,7 @@ namespace MQTTnet.Server
            return context;
        }

-        private async Task<MqttClientConnection> CreateConnectionAsync(IMqttChannelAdapter channelAdapter, MqttConnectPacket connectPacket)
+        private async Task<MqttClientConnection> CreateConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
        {
            await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false);
            try
@@ -345,7 +345,7 @@ namespace MQTTnet.Server

                if (session == null)
                {
-                    session = new MqttClientSession(connectPacket.ClientId, _eventDispatcher, _options, _logger);
+                    session = new MqttClientSession(connectPacket.ClientId, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _logger);
                    _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
                }

@@ -362,7 +362,7 @@ namespace MQTTnet.Server
            }
        }

-        private async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(MqttClientConnection sender, MqttApplicationMessage applicationMessage)
+        private async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)
        {
            var interceptor = _options.ApplicationMessageInterceptor;
            if (interceptor == null)
@@ -370,13 +370,22 @@ namespace MQTTnet.Server
                return null;
            }

-            var senderClientId = sender?.ClientId;
-            if (sender == null)
+            string senderClientId;
+            IDictionary<object, object> sessionItems;
+
+            var messageIsFromServer = senderConnection == null;
+            if (messageIsFromServer)
            {
                senderClientId = _options.ClientId;
+                sessionItems = _serverSessionItems;
+            }
+            else
+            {
+                senderClientId = senderConnection.ClientId;
+                sessionItems = senderConnection.Session.Items;
            }

-            var interceptorContext = new MqttApplicationMessageInterceptorContext(senderClientId, applicationMessage);
+            var interceptorContext = new MqttApplicationMessageInterceptorContext(senderClientId, sessionItems, applicationMessage);
            await interceptor.InterceptApplicationMessagePublishAsync(interceptorContext).ConfigureAwait(false);
            return interceptorContext;
        }
@@ -9,21 +9,24 @@ namespace MQTTnet.Server
 {
    public class MqttClientSubscriptionsManager
    {
-        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
-        private readonly IMqttServerOptions _options;
+        private readonly Dictionary<string, TopicFilter> _subscriptions = new Dictionary<string, TopicFilter>();
+        private readonly MqttClientSession _clientSession;
+        private readonly IMqttServerOptions _serverOptions;
        private readonly MqttServerEventDispatcher _eventDispatcher;
-        private readonly string _clientId;

-        public MqttClientSubscriptionsManager(string clientId, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions options)
+        public MqttClientSubscriptionsManager(MqttClientSession clientSession, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions)
        {
-            _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
-            _options = options ?? throw new ArgumentNullException(nameof(options));
+            _clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
+
+            // TODO: Consider removing the server options here and build a new class "ISubscriptionInterceptor" and just pass it. The instance is generated in the root server class upon start.
+            _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
            _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
        }

-        public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket)
+        public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket, MqttConnectPacket connectPacket)
        {
            if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
+            if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket));

            var result = new MqttClientSubscribeResult
            {
@@ -61,10 +64,10 @@ namespace MQTTnet.Server
                {
                    lock (_subscriptions)
                    {
-                        _subscriptions[finalTopicFilter.Topic] = finalTopicFilter.QualityOfServiceLevel;
+                        _subscriptions[finalTopicFilter.Topic] = finalTopicFilter;
                    }

-                    await _eventDispatcher.HandleClientSubscribedTopicAsync(_clientId, finalTopicFilter).ConfigureAwait(false);
+                    await _eventDispatcher.HandleClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false);
                }
            }

@@ -73,6 +76,8 @@ namespace MQTTnet.Server

        public async Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
        {
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
            foreach (var topicFilter in topicFilters)
            {
                var interceptorContext = await InterceptSubscribeAsync(topicFilter).ConfigureAwait(false);
@@ -85,10 +90,10 @@ namespace MQTTnet.Server
                {
                    lock (_subscriptions)
                    {
-                        _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                        _subscriptions[topicFilter.Topic] = topicFilter;
                    }

-                    await _eventDispatcher.HandleClientSubscribedTopicAsync(_clientId, topicFilter).ConfigureAwait(false);
+                    await _eventDispatcher.HandleClientSubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
                }
            }
        }
@@ -119,7 +124,7 @@ namespace MQTTnet.Server

            foreach (var topicFilter in unsubscribePacket.TopicFilters)
            {
-                await _eventDispatcher.HandleClientUnsubscribedTopicAsync(_clientId, topicFilter).ConfigureAwait(false);
+                await _eventDispatcher.HandleClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
            }
            
            return unsubAckPacket;
@@ -153,7 +158,7 @@ namespace MQTTnet.Server
                        continue;
                    }

-                    qosLevels.Add(subscription.Value);
+                    qosLevels.Add(subscription.Value.QualityOfServiceLevel);
                }
            }

@@ -192,10 +197,10 @@ namespace MQTTnet.Server

        private async Task<MqttSubscriptionInterceptorContext> InterceptSubscribeAsync(TopicFilter topicFilter)
        {
-            var context = new MqttSubscriptionInterceptorContext(_clientId, topicFilter);
-            if (_options.SubscriptionInterceptor != null)
+            var context = new MqttSubscriptionInterceptorContext(_clientSession.ClientId, topicFilter, _clientSession.Items);
+            if (_serverOptions.SubscriptionInterceptor != null)
            {
-                await _options.SubscriptionInterceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false);
+                await _serverOptions.SubscriptionInterceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false);
            }

            return context;
@@ -14,46 +14,15 @@ namespace MQTTnet.Server
        private readonly MqttConnectPacket _connectPacket;
        private readonly IMqttChannelAdapter _clientAdapter;

-        public MqttConnectionValidatorContext(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter)
+        public MqttConnectionValidatorContext(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter, IDictionary<object, object> sessionItems)
        {
-            _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
+            _connectPacket = connectPacket;
            _clientAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter));
+            SessionItems = sessionItems;
        }

        public string ClientId => _connectPacket.ClientId;

-        public string Username => _connectPacket.Username;
-
-        public byte[] RawPassword => _connectPacket.Password;
-
-        public string Password => Encoding.UTF8.GetString(RawPassword ?? new byte[0]);
-
-        public MqttApplicationMessage WillMessage => _connectPacket.WillMessage;
-
-        public bool CleanSession => _connectPacket.CleanSession;
-
-        public ushort KeepAlivePeriod => _connectPacket.KeepAlivePeriod;
-
-        public List<MqttUserProperty> UserProperties => _connectPacket.Properties?.UserProperties;
-
-        public byte[] AuthenticationData => _connectPacket.Properties?.AuthenticationData;
-
-        public string AuthenticationMethod => _connectPacket.Properties?.AuthenticationMethod;
-
-        public uint? MaximumPacketSize => _connectPacket.Properties?.MaximumPacketSize;
-
-        public ushort? ReceiveMaximum => _connectPacket.Properties?.ReceiveMaximum;
-
-        public ushort? TopicAliasMaximum => _connectPacket.Properties?.TopicAliasMaximum;
-
-        public bool? RequestProblemInformation => _connectPacket.Properties?.RequestProblemInformation;
-
-        public bool? RequestResponseInformation => _connectPacket.Properties?.RequestResponseInformation;
-
-        public uint? SessionExpiryInterval => _connectPacket.Properties?.SessionExpiryInterval;
-
-        public uint? WillDelayInterval => _connectPacket.Properties?.WillDelayInterval;
-
        public string Endpoint => _clientAdapter.Endpoint;

        public bool IsSecureConnection => _clientAdapter.IsSecureConnection;
@@ -62,6 +31,43 @@ namespace MQTTnet.Server

        public MqttProtocolVersion ProtocolVersion => _clientAdapter.PacketFormatterAdapter.ProtocolVersion;

+        public string Username => _connectPacket?.Username;
+
+        public byte[] RawPassword => _connectPacket?.Password;
+
+        public string Password => Encoding.UTF8.GetString(RawPassword ?? new byte[0]);
+
+        public MqttApplicationMessage WillMessage => _connectPacket?.WillMessage;
+
+        public bool? CleanSession => _connectPacket?.CleanSession;
+
+        public ushort? KeepAlivePeriod => _connectPacket?.KeepAlivePeriod;
+
+        public List<MqttUserProperty> UserProperties => _connectPacket?.Properties?.UserProperties;
+
+        public byte[] AuthenticationData => _connectPacket?.Properties?.AuthenticationData;
+
+        public string AuthenticationMethod => _connectPacket?.Properties?.AuthenticationMethod;
+
+        public uint? MaximumPacketSize => _connectPacket?.Properties?.MaximumPacketSize;
+
+        public ushort? ReceiveMaximum => _connectPacket?.Properties?.ReceiveMaximum;
+
+        public ushort? TopicAliasMaximum => _connectPacket?.Properties?.TopicAliasMaximum;
+
+        public bool? RequestProblemInformation => _connectPacket?.Properties?.RequestProblemInformation;
+
+        public bool? RequestResponseInformation => _connectPacket?.Properties?.RequestResponseInformation;
+
+        public uint? SessionExpiryInterval => _connectPacket?.Properties?.SessionExpiryInterval;
+
+        public uint? WillDelayInterval => _connectPacket?.Properties?.WillDelayInterval;
+
+        /// <summary>
+        /// Gets or sets a key/value collection that can be used to share data within the scope of this session.
+        /// </summary>
+        public IDictionary<object, object> SessionItems { get; }
+
        /// <summary>
        /// This is used for MQTTv3 only.
        /// </summary>
@@ -0,0 +1,7 @@
+namespace MQTTnet.Server
+{
+    public class MqttServerCertificateCredentials : IMqttServerCertificateCredentials
+    {
+        public string Password { get; set; }
+    }
+}
@@ -1,5 +1,6 @@
 using System;
 using System.Net;
+using System.Net.Security;
 using System.Security.Authentication;

 namespace MQTTnet.Server
@@ -81,9 +82,10 @@ namespace MQTTnet.Server
            return this;
        }

-        public MqttServerOptionsBuilder WithEncryptionCertificate(byte[] value)
+        public MqttServerOptionsBuilder WithEncryptionCertificate(byte[] value, IMqttServerCertificateCredentials credentials = null)
        {
            _options.TlsEndpointOptions.Certificate = value;
+            _options.TlsEndpointOptions.CertificateCredentials = credentials;
            return this;
        }

@@ -93,11 +95,29 @@ namespace MQTTnet.Server
            return this;
        }

+#if !WINDOWS_UWP
+        public MqttServerOptionsBuilder WithClientCertificate(RemoteCertificateValidationCallback validationCallback = null, bool checkCertificateRevocation = false)
+        {
+            _options.TlsEndpointOptions.ClientCertificateRequired = true;
+            _options.TlsEndpointOptions.CheckCertificateRevocation = checkCertificateRevocation;
+            _options.TlsEndpointOptions.RemoteCertificateValidationCallback = validationCallback;
+            return this;
+        }
+#endif
+
        public MqttServerOptionsBuilder WithoutEncryptedEndpoint()
        {
            _options.TlsEndpointOptions.IsEnabled = false;
            return this;
        }
+
+#if !WINDOWS_UWP
+        public MqttServerOptionsBuilder WithRemoteCertificateValidationCallback(RemoteCertificateValidationCallback value)
+        {
+            _options.TlsEndpointOptions.RemoteCertificateValidationCallback = value;
+            return this;
+        }
+#endif
        
        public MqttServerOptionsBuilder WithStorage(IMqttServerStorage value)
        {
@@ -19,5 +19,10 @@ namespace MQTTnet.Server
        public IPAddress BoundInterNetworkAddress { get; set; } = IPAddress.Any;

        public IPAddress BoundInterNetworkV6Address { get; set; } = IPAddress.IPv6Any;
+
+        /// <summary>
+        /// This requires admin permissions on Linux.
+        /// </summary>
+        public bool ReuseAddress { get; set; }
    }
 }
@@ -1,4 +1,5 @@
-using System.Security.Authentication;
+using System.Net.Security;
+using System.Security.Authentication;

 namespace MQTTnet.Server
 {
@@ -11,10 +12,15 @@ namespace MQTTnet.Server

        public byte[] Certificate { get; set; }

+        public IMqttServerCertificateCredentials CertificateCredentials { get; set; }
+
        public bool ClientCertificateRequired { get; set; }

        public bool CheckCertificateRevocation { get; set; }
-        
+
+#if !WINDOWS_UWP
+        public RemoteCertificateValidationCallback RemoteCertificateValidationCallback { get; set; }
+#endif
        public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12;
    }
 }
@@ -1,19 +1,25 @@
-using System;
+using System.Collections.Generic;

 namespace MQTTnet.Server
 {
    public class MqttSubscriptionInterceptorContext
    {
-        public MqttSubscriptionInterceptorContext(string clientId, TopicFilter topicFilter)
+        public MqttSubscriptionInterceptorContext(string clientId, TopicFilter topicFilter, IDictionary<object, object> sessionItems)
        {
            ClientId = clientId;
-            TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
+            TopicFilter = topicFilter;
+            SessionItems = sessionItems;
        }

        public string ClientId { get; }

        public TopicFilter TopicFilter { get; set; }
-        
+
+        /// <summary>
+        /// Gets or sets a key/value collection that can be used to share data within the scope of this session.
+        /// </summary>
+        public IDictionary<object, object> SessionItems { get; }
+
        public bool AcceptSubscription { get; set; } = true;

        public bool CloseConnection { get; set; }
@@ -2,12 +2,11 @@

  <PropertyGroup>
    <TargetFramework>netcoreapp2.1</TargetFramework>
-
    <IsPackable>false</IsPackable>
  </PropertyGroup>

  <ItemGroup>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.0.1" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.2.0" />
    <PackageReference Include="MSTest.TestAdapter" Version="1.4.0" />
    <PackageReference Include="MSTest.TestFramework" Version="1.4.0" />
  </ItemGroup>
@@ -4,13 +4,21 @@ namespace MQTTnet.AspNetCore.Tests.Mockups
 {
    public class DuplexPipeMockup : IDuplexPipe
    {
+        public DuplexPipeMockup()
+        {
+            var pool = new LimitedMemoryPool();
+            var pipeOptions = new PipeOptions(pool);
+            Receive = new Pipe(pipeOptions);
+            Send = new Pipe(pipeOptions);
+        }
+
        PipeReader IDuplexPipe.Input => Receive.Reader;

        PipeWriter IDuplexPipe.Output => Send.Writer;


-        public Pipe Receive { get; set; } = new Pipe();
+        public Pipe Receive { get; set; } 

-        public Pipe Send { get; set; } = new Pipe();
+        public Pipe Send { get; set; }
    }
 }
@@ -0,0 +1,18 @@
+using System.Buffers;
+
+namespace MQTTnet.AspNetCore.Tests.Mockups
+{
+    public class LimitedMemoryPool : MemoryPool<byte>
+    {
+        protected override void Dispose(bool disposing)
+        {
+        }
+
+        public override IMemoryOwner<byte> Rent(int minBufferSize = -1)
+        {
+            return new MemoryOwner(minBufferSize);
+        }
+
+        public override int MaxBufferSize { get; }
+    }
+}
@@ -0,0 +1,33 @@
+using System;
+using System.Buffers;
+
+namespace MQTTnet.AspNetCore.Tests.Mockups
+{
+    public class MemoryOwner : IMemoryOwner<byte>
+    {
+        private readonly byte[] _raw;
+
+        public MemoryOwner(int size)
+        {
+            if (size <= 0)
+            {
+                size = 1024;
+            }
+
+            if (size > 4096)
+            {
+                size = 4096;
+            }
+
+            _raw = ArrayPool<byte>.Shared.Rent(size);
+            Memory = _raw;
+        }
+
+        public void Dispose()
+        {
+            ArrayPool<byte>.Shared.Return(_raw);
+        }
+
+        public Memory<byte> Memory { get; }
+    }
+}
@@ -47,5 +47,21 @@ namespace MQTTnet.AspNetCore.Tests

            await Task.WhenAll(tasks).ConfigureAwait(false);
        }
+
+
+        [TestMethod]
+        public async Task TestLargePacket()
+        {
+            var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311);
+            var pipe = new DuplexPipeMockup();
+            var connection = new DefaultConnectionContext();
+            connection.Transport = pipe;
+            var ctx = new MqttConnectionContext(serializer, connection);
+
+            await ctx.SendPacketAsync(new MqttPublishPacket() { Payload = new byte[20_000] }, TimeSpan.Zero, CancellationToken.None).ConfigureAwait(false);
+
+            var readResult = await pipe.Send.Reader.ReadAsync();
+            Assert.IsTrue(readResult.Buffer.Length > 20000);
+        }
    }
 }
@@ -1,15 +1,14 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <OutputType>Exe</OutputType>
-    <DebugType>Full</DebugType>
    <TargetFramework>netcoreapp2.1</TargetFramework>
+    <IsPackable>false</IsPackable>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="MSTest.TestAdapter" Version="1.4.0" />
    <PackageReference Include="MSTest.TestFramework" Version="1.4.0" />
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.0.1" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.2.0" />
  </ItemGroup>

  <ItemGroup>
@@ -293,5 +293,52 @@ namespace MQTTnet.Tests.MQTTv5
                await server.StopAsync();
            }
        }
+
+        [TestMethod]
+        public async Task Publish_And_Receive_New_Properties()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+
+                var receiver = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
+                await receiver.SubscribeAsync("#");
+
+                MqttApplicationMessage receivedMessage = null;
+                receiver.UseApplicationMessageReceivedHandler(c =>
+                {
+                    receivedMessage = c.ApplicationMessage;
+                });
+
+                var sender = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
+
+                var applicationMessage = new MqttApplicationMessageBuilder()
+                    .WithTopic("Hello")
+                    .WithPayload("World")
+                    .WithAtMostOnceQoS()
+                    .WithUserProperty("x", "1")
+                    .WithUserProperty("y", "2")
+                    .WithResponseTopic("response")
+                    .WithContentType("text")
+                    .WithMessageExpiryInterval(50)
+                    .WithCorrelationData(new byte[12])
+                    .WithTopicAlias(2)
+                    .Build();
+
+                await sender.PublishAsync(applicationMessage);
+
+                await Task.Delay(500);
+
+                Assert.IsNotNull(receivedMessage);
+                Assert.AreEqual(applicationMessage.Topic, receivedMessage.Topic);
+                Assert.AreEqual(applicationMessage.TopicAlias, receivedMessage.TopicAlias);
+                Assert.AreEqual(applicationMessage.ContentType, receivedMessage.ContentType);
+                Assert.AreEqual(applicationMessage.ResponseTopic, receivedMessage.ResponseTopic);
+                Assert.AreEqual(applicationMessage.MessageExpiryInterval, receivedMessage.MessageExpiryInterval);
+                CollectionAssert.AreEqual(applicationMessage.CorrelationData, receivedMessage.CorrelationData);
+                CollectionAssert.AreEqual(applicationMessage.Payload, receivedMessage.Payload);
+                CollectionAssert.AreEqual(applicationMessage.UserProperties, receivedMessage.UserProperties);
+            }
+        }
    }
 }
@@ -1,4 +1,5 @@
-using System.Threading;
+using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Client;
@@ -108,5 +109,92 @@ namespace MQTTnet.Tests
                Assert.AreEqual(0, (await server.GetClientStatusAsync()).Count);
            }
        }
+        
+        [TestMethod]
+        public async Task Storage_Queue_Drains()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+                testEnvironment.IgnoreServerLogErrors = true;
+
+                var factory = new MqttFactory();
+
+                var server = await testEnvironment.StartServerAsync();
+
+                var managedClient = new ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetLogger().CreateChildLogger());
+                var clientOptions = new MqttClientOptionsBuilder()
+                    .WithTcpServer("localhost", testEnvironment.ServerPort);
+                var storage = new ManagedMqttClientTestStorage();
+
+                TaskCompletionSource<bool> connected = new TaskCompletionSource<bool>();
+                managedClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(e =>
+                {
+                    managedClient.ConnectedHandler = null;
+                    connected.SetResult(true);
+                });
+
+                await managedClient.StartAsync(new ManagedMqttClientOptionsBuilder()
+                    .WithClientOptions(clientOptions)
+                    .WithStorage(storage)
+                    .WithAutoReconnectDelay(System.TimeSpan.FromSeconds(5))
+                    .Build());
+
+                await connected.Task;
+
+                await testEnvironment.Server.StopAsync();
+
+                await managedClient.PublishAsync(new MqttApplicationMessage { Topic = "1" });
+
+                //Message should have been added to the storage queue in PublishAsync,
+                //and we are awaiting PublishAsync, so the message should already be
+                //in storage at this point (i.e. no waiting).
+                Assert.AreEqual(1, storage.GetMessageCount());
+
+                connected = new TaskCompletionSource<bool>();
+                managedClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(e =>
+                {
+                    managedClient.ConnectedHandler = null;
+                    connected.SetResult(true);
+                });
+
+                await testEnvironment.Server.StartAsync(new MqttServerOptionsBuilder()
+                    .WithDefaultEndpointPort(testEnvironment.ServerPort).Build());
+
+                await connected.Task;
+
+                //Wait 500ms here so the client has time to publish the queued message
+                await Task.Delay(500);
+
+                Assert.AreEqual(0, storage.GetMessageCount());
+
+                await managedClient.StopAsync();
+            }
+        }
    }
-}
+    
+    public class ManagedMqttClientTestStorage : IManagedMqttClientStorage
+    {
+        private IList<ManagedMqttApplicationMessage> _messages = null;
+
+        public Task<IList<ManagedMqttApplicationMessage>> LoadQueuedMessagesAsync()
+        {
+            if (_messages == null)
+            {
+                _messages = new List<ManagedMqttApplicationMessage>();
+            }
+            return Task.FromResult(_messages);
+        }
+
+        public Task SaveQueuedMessagesAsync(IList<ManagedMqttApplicationMessage> messages)
+        {
+            _messages = messages;
+            return Task.FromResult(0);
+        }
+
+        public int GetMessageCount()
+        {
+            return _messages.Count;
+        }
+    }
+}
@@ -109,6 +109,83 @@ namespace MQTTnet.Tests
            }
        }

+        [TestMethod]
+        public async Task Reconnect_While_Server_Offline()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+
+                var server = await testEnvironment.StartServerAsync();
+                var client = await testEnvironment.ConnectClientAsync();
+
+                await Task.Delay(500);
+                Assert.IsTrue(client.IsConnected);
+
+                await server.StopAsync();
+                await Task.Delay(500);
+                Assert.IsFalse(client.IsConnected);
+
+                for (var i = 0; i < 5; i++)
+                {
+                    try
+                    {
+                        await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
+                        Assert.Fail("Must fail!");
+                    }
+                    catch
+                    {
+                    }
+                }
+
+                await server.StartAsync(new MqttServerOptionsBuilder().WithDefaultEndpointPort(testEnvironment.ServerPort).Build());
+                await Task.Delay(500);
+
+                await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
+                Assert.IsTrue(client.IsConnected);
+            }
+        }
+
+        [TestMethod]
+        public async Task Reconnect_From_Disconnected_Event()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+
+                var client = testEnvironment.CreateClient();
+
+                var tries = 0;
+                var maxTries = 3;
+
+                client.UseDisconnectedHandler(async e =>
+                {
+                    if (tries >= maxTries)
+                    {
+                        return;
+                    }
+
+                    Interlocked.Increment(ref tries);
+
+                    await Task.Delay(100);
+                    await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
+                });
+
+                try
+                {
+                    await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build());
+                    Assert.Fail("Must fail!");
+                }
+                catch
+                {
+                }
+                
+                SpinWait.SpinUntil(() => tries >= maxTries, 10000);
+
+                Assert.AreEqual(maxTries, tries);
+            }
+        }
+
        [TestMethod]
        public async Task PacketIdentifier_In_Publish_Result()
        {
@@ -158,6 +235,25 @@ namespace MQTTnet.Tests
            }
        }

+        [TestMethod]
+        public async Task Fire_Disconnected_Event_On_Server_Shutdown()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                var server = await testEnvironment.StartServerAsync();
+                var client = await testEnvironment.ConnectClientAsync();
+
+                var handlerFired = false;
+                client.UseDisconnectedHandler(e => handlerFired = true);
+
+                await server.StopAsync();
+
+                await Task.Delay(4000);
+
+                Assert.IsTrue(handlerFired);
+            }
+        }
+
        [TestMethod]
        public async Task Disconnect_Event_Contains_Exception()
        {
@@ -14,7 +14,7 @@ namespace MQTTnet.Tests
    public class MqttKeepAliveMonitor_Tests
    {
        [TestMethod]
-        public void KeepAlive_Timeout()
+        public async Task KeepAlive_Timeout()
        {
            var counter = 0;

@@ -31,13 +31,13 @@ namespace MQTTnet.Tests

            Assert.AreEqual(0, counter);

-            Thread.Sleep(2000); // Internally the keep alive timeout is multiplied with 1.5 as per protocol specification.
+            await Task.Delay(2000); // Internally the keep alive timeout is multiplied with 1.5 as per protocol specification.

            Assert.AreEqual(1, counter);
        }

        [TestMethod]
-        public void KeepAlive_NoTimeout()
+        public async Task KeepAlive_NoTimeout()
        {
            var counter = 0;

@@ -55,15 +55,15 @@ namespace MQTTnet.Tests
            Assert.AreEqual(0, counter);

            // Simulate traffic.
-            Thread.Sleep(1000); // Internally the keep alive timeout is multiplied with 1.5 as per protocol specification.
+            await Task.Delay(1000); // Internally the keep alive timeout is multiplied with 1.5 as per protocol specification.
            monitor.PacketReceived();
-            Thread.Sleep(1000);
+            await Task.Delay(1000);
            monitor.PacketReceived();
-            Thread.Sleep(1000);
+            await Task.Delay(1000);

            Assert.AreEqual(0, counter);

-            Thread.Sleep(2000);
+            await Task.Delay(2000);

            Assert.AreEqual(1, counter);
        }
@@ -8,6 +8,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Adapter;
 using MQTTnet.Formatter;
 using MQTTnet.Formatter.V3;
+using MQTTnet.Formatter.V5;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
@@ -258,6 +259,26 @@ namespace MQTTnet.Tests
            Assert.IsTrue(deserialized.Properties.UserProperties.Any(x => x.Name == "Foo"));
        }

+
+        [TestMethod]
+        public void SerializeV500_MqttPublishPacket_CorrelationData()
+        {
+            var data = "123456789";
+            var req = new MqttApplicationMessageBuilder()
+                      .WithTopic("Foo")
+                      .WithResponseTopic($"_")
+                      .WithCorrelationData(Guid.NewGuid().ToByteArray())
+                      .WithPayload(data)
+                      .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce)
+                      .Build();
+
+            var p = new MqttV500DataConverter().CreatePublishPacket(req);
+
+            var deserialized = Roundtrip(p, MqttProtocolVersion.V500);
+
+            Assert.IsTrue(p.Payload.SequenceEqual(deserialized.Payload));
+        }
+
        [TestMethod]
        public void DeserializeV311_MqttPublishPacket()
        {
@@ -1,4 +1,6 @@
-using Microsoft.VisualStudio.TestTools.UnitTesting;
+using System.Collections.Concurrent;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
@@ -10,14 +12,17 @@ namespace MQTTnet.Tests
    public class MqttSubscriptionsManager_Tests
    {
        [TestMethod]
-        public void MqttSubscriptionsManager_SubscribeSingleSuccess()
+        public async Task MqttSubscriptionsManager_SubscribeSingleSuccess()
        {
-            var sm = new MqttClientSubscriptionsManager("", new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());
+            var s = new MqttClientSession("", new ConcurrentDictionary<object, object>(),
+                new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions(), new TestLogger());
+
+            var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());

            var sp = new MqttSubscribePacket();
            sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());

-            sm.SubscribeAsync(sp).GetAwaiter().GetResult();
+            await sm.SubscribeAsync(sp, new MqttConnectPacket());

            var result = sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce);
            Assert.IsTrue(result.IsSubscribed);
@@ -25,14 +30,17 @@ namespace MQTTnet.Tests
        }

        [TestMethod]
-        public void MqttSubscriptionsManager_SubscribeDifferentQoSSuccess()
+        public async Task MqttSubscriptionsManager_SubscribeDifferentQoSSuccess()
        {
-            var sm = new MqttClientSubscriptionsManager("", new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());
+            var s = new MqttClientSession("", new ConcurrentDictionary<object, object>(),
+                new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions(), new TestLogger());
+
+            var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());

            var sp = new MqttSubscribePacket();
            sp.TopicFilters.Add(new TopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });

-            sm.SubscribeAsync(sp).GetAwaiter().GetResult();
+            await sm.SubscribeAsync(sp, new MqttConnectPacket());

            var result = sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.ExactlyOnce);
            Assert.IsTrue(result.IsSubscribed);
@@ -40,15 +48,18 @@ namespace MQTTnet.Tests
        }

        [TestMethod]
-        public void MqttSubscriptionsManager_SubscribeTwoTimesSuccess()
+        public async Task MqttSubscriptionsManager_SubscribeTwoTimesSuccess()
        {
-            var sm = new MqttClientSubscriptionsManager("", new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());
+            var s = new MqttClientSession("", new ConcurrentDictionary<object, object>(),
+                new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions(), new TestLogger());
+
+            var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());

            var sp = new MqttSubscribePacket();
            sp.TopicFilters.Add(new TopicFilter { Topic = "#", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
            sp.TopicFilters.Add(new TopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });

-            sm.SubscribeAsync(sp).GetAwaiter().GetResult();
+            await sm.SubscribeAsync(sp, new MqttConnectPacket());

            var result = sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.ExactlyOnce);
            Assert.IsTrue(result.IsSubscribed);
@@ -56,33 +67,39 @@ namespace MQTTnet.Tests
        }

        [TestMethod]
-        public void MqttSubscriptionsManager_SubscribeSingleNoSuccess()
+        public async Task MqttSubscriptionsManager_SubscribeSingleNoSuccess()
        {
-            var sm = new MqttClientSubscriptionsManager("", new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());
+            var s = new MqttClientSession("", new ConcurrentDictionary<object, object>(),
+                new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions(), new TestLogger());
+
+            var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());

            var sp = new MqttSubscribePacket();
            sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());

-            sm.SubscribeAsync(sp).GetAwaiter().GetResult();
+            await sm.SubscribeAsync(sp, new MqttConnectPacket());

            Assert.IsFalse(sm.CheckSubscriptions("A/B/X", MqttQualityOfServiceLevel.AtMostOnce).IsSubscribed);
        }

        [TestMethod]
-        public void MqttSubscriptionsManager_SubscribeAndUnsubscribeSingle()
+        public async Task MqttSubscriptionsManager_SubscribeAndUnsubscribeSingle()
        {
-            var sm = new MqttClientSubscriptionsManager("", new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());
+            var s = new MqttClientSession("", new ConcurrentDictionary<object, object>(),
+                new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions(), new TestLogger());
+
+            var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());

            var sp = new MqttSubscribePacket();
            sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());

-            sm.SubscribeAsync(sp).GetAwaiter().GetResult();
+            await sm.SubscribeAsync(sp, new MqttConnectPacket());

            Assert.IsTrue(sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce).IsSubscribed);

            var up = new MqttUnsubscribePacket();
            up.TopicFilters.Add("A/B/C");
-            sm.UnsubscribeAsync(up);
+            await sm.UnsubscribeAsync(up);

            Assert.IsFalse(sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce).IsSubscribed);
        }
@@ -8,6 +8,10 @@ using MQTTnet.Client.Receiving;
 using MQTTnet.Exceptions;
 using MQTTnet.Extensions.Rpc;
 using MQTTnet.Protocol;
+using MQTTnet.Client.Options;
+using MQTTnet.Formatter;
+using MQTTnet.Extensions.Rpc.Options;
+using MQTTnet.Extensions.Rpc.Options.TopicGeneration;

 namespace MQTTnet.Tests
 {
@@ -15,26 +19,39 @@ namespace MQTTnet.Tests
    public class RPC_Tests
    {
        [TestMethod]
-        public async Task Execute_Success()
+        public Task Execute_Success_With_QoS_0()
        {
-            using (var testEnvironment = new TestEnvironment())
-            {
-                await testEnvironment.StartServerAsync();
-                var responseSender = await testEnvironment.ConnectClientAsync();
-                await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping");
+            return Execute_Success(MqttQualityOfServiceLevel.AtMostOnce, MqttProtocolVersion.V311);
+        }

-                responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(async e =>
-                {
-                    await responseSender.PublishAsync(e.ApplicationMessage.Topic + "/response", "pong");
-                });
+        [TestMethod]
+        public Task Execute_Success_With_QoS_1()
+        {
+            return Execute_Success(MqttQualityOfServiceLevel.AtLeastOnce, MqttProtocolVersion.V311);
+        }

-                var requestSender = await testEnvironment.ConnectClientAsync();
+        [TestMethod]
+        public Task Execute_Success_With_QoS_2()
+        {
+            return Execute_Success(MqttQualityOfServiceLevel.ExactlyOnce, MqttProtocolVersion.V311);
+        }

-                var rpcClient = new MqttRpcClient(requestSender);
-                var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
+        [TestMethod]
+        public Task Execute_Success_With_QoS_0_MQTT_V5()
+        {
+            return Execute_Success(MqttQualityOfServiceLevel.AtMostOnce, MqttProtocolVersion.V500);
+        }

-                Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
-            }
+        [TestMethod]
+        public Task Execute_Success_With_QoS_1_MQTT_V5()
+        {
+            return Execute_Success(MqttQualityOfServiceLevel.AtLeastOnce, MqttProtocolVersion.V500);
+        }
+
+        [TestMethod]
+        public Task Execute_Success_With_QoS_2_MQTT_V5()
+        {
+            return Execute_Success(MqttQualityOfServiceLevel.ExactlyOnce, MqttProtocolVersion.V500);
        }

        [TestMethod]
@@ -47,9 +64,58 @@ namespace MQTTnet.Tests
                
                var requestSender = await testEnvironment.ConnectClientAsync();

-                var rpcClient = new MqttRpcClient(requestSender);
+                var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build());
                await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
            }
        }
+
+        [TestMethod]
+        [ExpectedException(typeof(MqttCommunicationTimedOutException))]
+        public async Task Execute_With_Custom_Topic_Names()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+
+                var requestSender = await testEnvironment.ConnectClientAsync();
+
+                var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().WithTopicGenerationStrategy(new TestTopicStrategy()) .Build());
+                await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
+            }
+        }
+
+        private async Task Execute_Success(MqttQualityOfServiceLevel qosLevel, MqttProtocolVersion protocolVersion)
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+                var responseSender = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(protocolVersion));
+                await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping");
+
+                responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(async e =>
+                {
+                    await responseSender.PublishAsync(e.ApplicationMessage.Topic + "/response", "pong");
+                });
+
+                var requestSender = await testEnvironment.ConnectClientAsync();
+
+                var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build());
+                var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", qosLevel);
+
+                Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
+            }
+        }
+
+        private class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
+        {
+            public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
+            {
+                return new MqttRpcTopicPair
+                {
+                    RequestTopic = "a",
+                    ResponseTopic = "b"
+                };
+            }
+        }
    }
 }
@@ -1063,7 +1063,9 @@ namespace MQTTnet.Tests
                // forever. This is security related.
                var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                await client.ConnectAsync("localhost", testEnvironment.ServerPort);
-                await client.SendAsync(Encoding.UTF8.GetBytes("Garbage"), SocketFlags.None);
+                                
+                var buffer = Encoding.UTF8.GetBytes("Garbage");
+                client.Send(buffer, buffer.Length, SocketFlags.None);

                await Task.Delay(TimeSpan.FromSeconds(3));

@@ -0,0 +1,61 @@
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Client;
+using MQTTnet.Client.Subscribing;
+using MQTTnet.Server;
+using MQTTnet.Tests.Mockups;
+
+namespace MQTTnet.Tests
+{
+    [TestClass]
+    public class Session_Tests
+    {
+        [TestMethod]
+        public async Task Set_Session_Item()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                var serverOptions = new MqttServerOptionsBuilder()
+                    .WithConnectionValidator(delegate (MqttConnectionValidatorContext context)
+                    {
+                        // Don't validate anything. Just set some session items.
+                        context.SessionItems["can_subscribe_x"] = true;
+                        context.SessionItems["default_payload"] = "Hello World";
+                    })
+                    .WithSubscriptionInterceptor(delegate (MqttSubscriptionInterceptorContext context)
+                    {
+                        if (context.TopicFilter.Topic == "x")
+                        {
+                            context.AcceptSubscription = context.SessionItems["can_subscribe_x"] as bool? == true;
+                        }
+                    })
+                    .WithApplicationMessageInterceptor(delegate (MqttApplicationMessageInterceptorContext context)
+                    {
+                        context.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(context.SessionItems["default_payload"] as string);
+                    });
+
+                await testEnvironment.StartServerAsync(serverOptions);
+
+                string receivedPayload = null;
+
+                var client = await testEnvironment.ConnectClientAsync();
+                client.UseApplicationMessageReceivedHandler(delegate(MqttApplicationMessageReceivedEventArgs args)
+                {
+                    receivedPayload = args.ApplicationMessage.ConvertPayloadToString();
+                });
+
+                var subscribeResult = await client.SubscribeAsync("x");
+
+                Assert.AreEqual(MqttClientSubscribeResultCode.GrantedQoS0, subscribeResult.Items[0].ResultCode);
+
+                var client2 = await testEnvironment.ConnectClientAsync();
+                await client2.PublishAsync("x");
+
+                await Task.Delay(1000);
+
+                Assert.AreEqual("Hello World", receivedPayload);
+            }
+        }
+    }
+}
Author	SHA1	Message	Date
Christian	8ecf658059	Update MQTTnet.nuspec	2019-12-28 14:05:24 +01:00
Christian Kratky	851db2e808	Merge remote-tracking branch 'origin/master'	2019-10-03 15:23:42 +02:00
Christian Kratky	073cd5243f	Update docs. Add UnitTests.	2019-10-03 15:23:35 +02:00
SeppPenner	2918be18c6	Added automatic labels to the issue templates.	2019-09-05 10:37:55 +02:00
Christian Kratky	9302e80d73	Merge branch 'develop'	2019-08-18 21:00:40 +02:00
Christian Kratky	cf6f0ece0e	Add test execution in build script.	2019-08-18 20:56:26 +02:00
Christian Kratky	50d65f2fed	Update nugets.	2019-08-14 23:06:45 +02:00
Christian Kratky	4614df7fa5	Update docs.	2019-08-14 22:57:17 +02:00
Christian	22aedcba09	Merge pull request #744 from JanEggers/pr/fix#743 fixed #743	2019-08-14 22:53:55 +02:00
Christian Kratky	2767d897a4	Fix serialization issue of MQTTv5 PubRec packet.	2019-08-14 22:45:15 +02:00
JanEggers	a4420063c0	fixed #743	2019-08-14 22:29:51 +02:00
Christian Kratky	2320352124	Update docs.	2019-08-13 20:09:21 +02:00
Christian Kratky	c0b4155ab4	Update docs.	2019-08-13 20:07:14 +02:00
Christian Kratky	b09f31d654	Fix TCP server adpater when using empty password.	2019-08-13 20:07:10 +02:00
Christian Kratky	30c33d46f9	Merge remote-tracking branch 'origin/develop' into develop	2019-08-13 19:56:36 +02:00
Christian	651b79a124	Merge pull request #740 from cslutgen/chain Fix Client SslStream CRL verification	2019-08-13 19:56:26 +02:00
Christian Kratky	2e3cf54f11	Add support for different strategies when generating RPC topics.	2019-08-13 19:53:55 +02:00
Craig Lutgen	d7948afd98	Fix client SslStream certificate chain CRL validation. The logic was inverse and CRLs were never checked. Add a missing dispose in managed client.	2019-08-12 17:34:05 -05:00
cslutgen	4cefa1dfa9	Merge pull request #2 from chkr1011/develop Develop	2019-08-12 17:28:31 -05:00
Christian Kratky	0cfb8b4020	Remove not async task handling.	2019-08-11 22:25:05 +02:00
Christian Kratky	9ac726b271	Revert "Remove dedicated UWP builds and code." This reverts commit `6d0e2c2022`.	2019-08-11 22:19:49 +02:00
Christian Kratky	232605304e	Revert "Update nugets." This reverts commit `6c53677e7d`.	2019-08-11 22:19:29 +02:00
Christian Kratky	6c53677e7d	Update nugets.	2019-08-11 22:10:49 +02:00
Christian Kratky	6d0e2c2022	Remove dedicated UWP builds and code.	2019-08-11 22:07:57 +02:00
Christian Kratky	8ad1fb2871	Refactor server certificate classes (breaking change).	2019-08-11 21:46:13 +02:00
Christian Kratky	4a7ff9e9ed	Add TLS certificate password support for MQTTnet.Server.	2019-08-11 21:45:41 +02:00
Christian Kratky	f54598c201	Merge branch 'develop'	2019-08-11 16:33:40 +02:00
Christian Kratky	2300e68fff	Add UnitTests for RPC.	2019-08-11 10:41:36 +02:00
Christian Kratky	6996824157	Merge branch 'master' into develop	2019-08-11 10:17:51 +02:00
Christian Kratky	524f695829	Update docs.	2019-08-09 22:21:42 +02:00
Christian	72faa067cc	Merge pull request #729 from cslutgen/clientcerts Add Server certificate password and client certificate builders	2019-08-09 22:07:26 +02:00
SeppPenner	01aa75471c	Updated description of issue templates.	2019-08-09 11:20:28 +02:00
HansM	048206eb1e	Fixed my spelling mistake from before.	2019-08-09 11:14:52 +02:00
HansM	ad5b0caa4d	Fixed spelling issue in feature request issue template.	2019-08-09 11:13:40 +02:00
HansM	0f14ddcf79	Updated bug report issue template.	2019-08-09 11:13:06 +02:00
Craig Lutgen	59d2a8e551	Changed server crentials property name from password to certificateCredentials	2019-08-08 13:35:14 -05:00
HansM	8607ceb3b7	Fixed feature request issue template.	2019-08-02 08:47:09 +02:00
HansM	b82e5a48a4	Fixed custom issue template.	2019-08-02 08:46:35 +02:00
HansM	30c5147c62	Fixed bug reports issue template.	2019-08-02 08:46:12 +02:00
Craig Lutgen	9d63500f95	Fix build issue with UWP	2019-07-31 13:27:33 -05:00
Craig Lutgen	a03bdc7616	Expose server certificate password and client certificate options to MqttServerOptionsBuilder	2019-07-31 13:04:03 -05:00
Christian Kratky	413daee551	Merge branch 'master' into develop	2019-07-30 20:35:48 +02:00
Christian Kratky	f3ea7008e5	Merge remote-tracking branch 'origin/develop' into develop	2019-07-30 20:35:39 +02:00
Christian Kratky	8bfbbd2b29	Refactoring.	2019-07-30 20:35:33 +02:00
Christian Kratky	ab8c7df0b9	Add unit Tests.	2019-07-30 20:35:22 +02:00
Christian Kratky	975eb60e90	Fix memory leak when SSL is not working properly.	2019-07-30 20:02:46 +02:00
Christian Kratky	6076151222	Update docs.	2019-07-30 20:01:57 +02:00
Christian Kratky	9cc96f1888	Refactor log messages.	2019-07-30 20:01:45 +02:00
HansM	7aa4d5159c	Fixed styling in feature request template.	2019-07-30 09:56:00 +02:00
Christian	9b021c4093	Update README.md	2019-07-27 09:57:36 +02:00
Christian	54c303369e	Update README.md	2019-07-27 09:56:59 +02:00
Christian	8f30042e49	Update README.md	2019-07-27 09:56:46 +02:00
Christian	4c33270d26	Merge pull request #718 from PaulFake/patch-3 Fixed AsyncLock::WaitAsync cancellation bug	2019-07-26 21:23:29 +02:00
Paul Fake	5c40577bec	Fixed AsyncLock::WaitAsync cancellation bug There isn't any code in MQTTnet that actually uses a cancellation token in WaitAsync, so this is more of a preventative thing than a bug fix. The original code just checks if the task was completed, not whether it was cancelled. If the process is cancelled immediately before the call to WaitAsync, it'll return as normal (https://referencesource.microsoft.com/#mscorlib/system/threading/SemaphoreSlim.cs,612) rather than throw a cancellation exception. This change will ensure we only return the releaser if the wait actually ran to completion rather than exited early due to cancellation. I've tested this, and it properly throws a cancellation exception later on.	2019-07-23 16:20:36 -04:00
Christian Kratky	3d404f9755	Fixed build.	2019-07-23 22:08:51 +02:00
Christian Kratky	1b5e0197e4	Update docs.	2019-07-23 21:56:14 +02:00
Christian	1dab729d84	Merge pull request #717 from PaulFake/patch-2 Storage queue drain test	2019-07-23 21:54:48 +02:00
Christian	3eac5bcaf3	Merge pull request #709 from PaulFake/patch-1 Fixed storage queue race condition	2019-07-23 21:54:21 +02:00
Paul Fake	8eb23f0f1d	Storage queue drain test It's a unit test to ensure the storage queue is drained.	2019-07-23 15:45:50 -04:00
Christian	1fd1a12fce	Merge pull request #715 from mseyock/change-icon Remove shadow and made text bold	2019-07-23 19:59:32 +02:00
Matthias Seyock	349d506b37	Remove shadow and made text bold	2019-07-23 16:03:18 +02:00
Christian	a981b05aa7	Merge pull request #712 from mseyock/change-icon fix typo in icon	2019-07-19 22:12:59 +02:00
Matthias Seyock	654f06d20d	fix typo in icon	2019-07-19 20:34:39 +02:00
Christian	c92cd10499	Merge pull request #711 from mseyock/change-icon Replace icon	2019-07-19 19:33:32 +02:00
Christian Kratky	ed06ea6bc9	Add version information to MQTTnet.Server.	2019-07-19 19:32:33 +02:00
Matthias Seyock	3d531151c6	Replace icon	2019-07-19 16:52:58 +02:00
Paul Fake	e1590fc0c8	Fixed storage queue race condition Sometimes, TryPublishQueuedMessageAsync would try to remove a message from the storage queue before PublishAsync added it to the storage queue, resulting in a message being stuck in the storage queue forever. Switched the message queue lock to an async lock and synchronized the storage queue updates with the message queue updates.	2019-07-17 16:55:56 -04:00
Christian Kratky	dbb227406b	Added UnitTests.	2019-07-17 21:58:37 +02:00
Christian	b891e1d1f4	Update issue templates	2019-07-17 19:45:08 +02:00
Christian	1d697e3b87	Merge pull request #702 from rudacs/clientCertificate-is-null X509Certificate convert to X509Certificate2 in ClientCertificate	2019-07-15 19:18:03 +02:00
Rudá Cunha	128484c55a	X509Certificate convert to X509Certificate2 in ClientCertificate	2019-07-15 10:46:45 -03:00
Christian Kratky	ff22c34bb2	Add first classes for persisted session support.	2019-07-12 20:36:46 +02:00
Christian Kratky	e7ba3f76dc	Add options builder methods for remote certificate validation callback.	2019-07-11 23:03:56 +02:00
Christian	1ae3324581	Merge pull request #697 from rudacs/add-server-remove-validation-callback Add remote certificate validation callback on server	2019-07-11 22:56:36 +02:00
Rudá Cunha	ba8ce15ab4	Add remote certificate validation callback on server	2019-07-11 16:28:09 -03:00
Christian Kratky	87b4ce5185	Extend build and upload script to publish symbols as well.	2019-07-11 21:28:07 +02:00
Christian Kratky	4eb2b77138	Fix wrong NoDelay usage in server implementation.	2019-07-11 21:11:11 +02:00
Christian Kratky	70ee49f3c6	Merge remote-tracking branch 'origin/develop' into develop	2019-07-09 22:42:50 +02:00
Christian Kratky	f223a26748	Update docs.	2019-07-09 22:42:44 +02:00
Christian	4574fdc951	Merge pull request #693 from JanEggers/pr/fix691 fixed #691	2019-07-09 22:39:48 +02:00
Jan Eggers	4f66614cdd	fixed #691	2019-07-09 14:15:46 +02:00
Christian Kratky	6608172719	MQTTnet.Server: Extend messages API.	2019-07-07 21:26:07 +02:00
Christian Kratky	6c8db47e25	Remove dedicated values from contexts and only provide session items.	2019-07-05 18:28:18 +02:00
Christian Kratky	3eb5e82d10	Expose connect packet in application message interceptor and subscription interceptor.	2019-07-04 23:06:37 +02:00
Christian Kratky	8563230f9c	Update docs.	2019-07-04 20:38:01 +02:00
Christian Kratky	019f428f0a	MQTTnet.Server: Add HTTP API for publishing of messages.	2019-07-04 20:27:48 +02:00
Christian Kratky	1847af13d2	Merge branch 'develop'	2019-06-28 10:20:51 +02:00
Christian Kratky	8916df8d7b	Move ReuseAddress to server options.	2019-06-27 15:20:27 +02:00