Merge branch 'develop'

Add UnitTests for RPC.
Merge branch 'master' into develop
2026-10-15 00:15:15 +08:00 · 2019-08-11 16:33:40 +02:00 · 2019-08-11 10:41:36 +02:00 · 2019-08-11 10:17:51 +02:00 · 2019-08-09 22:21:42 +02:00 · 2019-08-09 22:07:26 +02:00
57 changed files with 1149 additions and 223 deletions
@@ -0,0 +1,46 @@
+---
+name: Bug report
+about: Create a report to help us improve.
+title: ''
+labels: ''
+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: ''
+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: ''
+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,22 +11,16 @@
    <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 issues in MQTTv5 message encoding and decoding.
-* [Core] Added extension method to allow usage of _WebSocket4Net_ in clients to fix issues with AWS and Xamarin.
-* [Core] Fixed usage of wrong data type for passwords (string -> byte[]).
-* [Core] Fixed an _ObjectDisposedException_ when sending data using a WebSocket channel.
-* [Core] Performance optimizations.
-* [Client] Added support for extended authentication exchange.
-* [Client] Exposed MQTTv5 CONNACK values to client.
-* [Client] Added _MqttClientSubscribeOptionsBuilder_.
-* [Client] The disconnected handler is now executed in a new task to prevent deadlocks when reconnecting etc. (thanks to @lizziebeans).
-* [Client] Converted option _DualMode_ into nullable boolean to preserve original value and avoid exceptions in IPv4 only networks (thanks to @lavaflo).
-* [Server] Exposed _ClientCertificateRequired_ and _CheckCertificateRevocation_ at TLS options.
-* [Server] Exposed client certificate at client connection validator.
-* [Server] The subscription interceptor now supports altering the entire topic filter.
-* [Server] Exposed more properties to the connection validator context.
-* [MQTTnet.Server] Added authentication support via Python script file.
-* [MQTTnet.Server] Migrated the result of connection validations to _ReasonCode_ (MQTTv5) instead of _ReturnCode_ (MQTTv3 only) (BREAKING CHANGE!).
+* [Core] Nuget packages with symbols are now also published to improve debugging.
+* [Core] Improve task handling (thanks to @mwinterb)
+* [ManagedClient] Fix a race condition in the message storage (thanks to @PaulFake).
+* [Server] Added items dictionary to client session in order to share data across interceptors as along as the session exists.
+* [Server] Exposed CONNECT packet properties in Application Message and Subscription interceptor.
+* [Server] Fixed: Sending Large packets with AspnetCore based connection throws System.ArgumentException.
+* [Server] Fixed wrong usage of socket option _NoDelay_.
+* [Server] Added remote certificate validation callback (thanks to @rudacs).
+* [Server] Add support for certificate passwords (thanks to @cslutgen).
+* [MQTTnet.Server] Added REST API for publishing basic messages.
    </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> 
@@ -43,6 +43,36 @@ Write-Host
 &$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 &$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
+
+Copy-Item MQTTnet.AspNetCore.nuspec -Destination MQTTnet.AspNetCore.nuspec.old -Force
+(Get-Content MQTTnet.AspNetCore.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.AspNetCore.nuspec
+Copy-Item MQTTnet.Extensions.Rpc.nuspec -Destination MQTTnet.Extensions.Rpc.nuspec.old -Force
+(Get-Content MQTTnet.Extensions.Rpc.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.Extensions.Rpc.nuspec
+Copy-Item MQTTnet.Extensions.ManagedClient.nuspec -Destination MQTTnet.Extensions.ManagedClient.nuspec.old -Force
+(Get-Content MQTTnet.Extensions.ManagedClient.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.Extensions.ManagedClient.nuspec
+Copy-Item MQTTnet.Extensions.WebSocket4Net.nuspec -Destination MQTTnet.Extensions.WebSocket4Net.nuspec.old -Force
+(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 -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
+Move-Item MQTTnet.Extensions.ManagedClient.nuspec.old -Destination MQTTnet.Extensions.ManagedClient.nuspec -Force
+Move-Item MQTTnet.Extensions.WebSocket4Net.nuspec.old -Destination MQTTnet.Extensions.WebSocket4Net.nuspec -Force
+
+Remove-Item "nuget.exe" -Force -Recurse -ErrorAction SilentlyContinue
+
 ####################################################################

 # Build MQTTnet.Server Portable
@@ -87,34 +117,4 @@ If(Test-path $destination) {Remove-item $destination}
 Add-Type -assembly "system.io.compression.filesystem"
 [io.compression.zipfile]::CreateFromDirectory($source, $destination) 

-####################################################################
-
-# 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
-
-Copy-Item MQTTnet.AspNetCore.nuspec -Destination MQTTnet.AspNetCore.nuspec.old -Force
-(Get-Content MQTTnet.AspNetCore.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.AspNetCore.nuspec
-Copy-Item MQTTnet.Extensions.Rpc.nuspec -Destination MQTTnet.Extensions.Rpc.nuspec.old -Force
-(Get-Content MQTTnet.Extensions.Rpc.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.Extensions.Rpc.nuspec
-Copy-Item MQTTnet.Extensions.ManagedClient.nuspec -Destination MQTTnet.Extensions.ManagedClient.nuspec.old -Force
-(Get-Content MQTTnet.Extensions.ManagedClient.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.Extensions.ManagedClient.nuspec
-Copy-Item MQTTnet.Extensions.WebSocket4Net.nuspec -Destination MQTTnet.Extensions.WebSocket4Net.nuspec.old -Force
-(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
-
-Move-Item MQTTnet.AspNetCore.nuspec.old -Destination MQTTnet.AspNetCore.nuspec -Force
-Move-Item MQTTnet.Extensions.Rpc.nuspec.old -Destination MQTTnet.Extensions.Rpc.nuspec -Force
-Move-Item MQTTnet.Extensions.ManagedClient.nuspec.old -Destination MQTTnet.Extensions.ManagedClient.nuspec -Force
-Move-Item MQTTnet.Extensions.WebSocket4Net.nuspec.old -Destination MQTTnet.Extensions.WebSocket4Net.nuspec -Force
-
-Remove-Item "nuget.exe" -Force -Recurse -ErrorAction SilentlyContinue
+####################################################################
@@ -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>
@@ -9,6 +9,7 @@
 [![BCH compliance](https://bettercodehub.com/edge/badge/chkr1011/MQTTnet?branch=master)](https://bettercodehub.com/)
 [![OpenCollective](https://opencollective.com/mqttnet/backers/badge.svg)](https://opencollective.com/mqttnet) 
 [![OpenCollective](https://opencollective.com/mqttnet/sponsors/badge.svg)](https://opencollective.com/mqttnet) [![Join the chat at https://gitter.im/MQTTnet/community](https://badges.gitter.im/MQTTnet/community.svg)](https://gitter.im/MQTTnet/community?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://raw.githubusercontent.com/chkr1011/MQTTnet/master/LICENSE)

 # MQTTnet

@@ -104,6 +105,8 @@ This library is used in the following projects:
 * MQTT Tester (MQTT client test app for [Android](https://play.google.com/store/apps/details?id=com.liveowl.mqtttester) and [iOS](https://itunes.apple.com/us/app/mqtt-tester/id1278621826?mt=8))
 * Wirehome.Core (Open Source Home Automation system for .NET Core, <https://github.com/chkr1011/Wirehome.Core>)

+Further projects using this project can be found under https://github.com/chkr1011/MQTTnet/network/dependents.
+
 If you use this library and want to see your project here please create a pull request.

 ## License
@@ -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
            {
@@ -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)
@@ -362,7 +364,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 +379,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 +388,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 +410,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 +535,4 @@ namespace MQTTnet.Extensions.ManagedClient
            _connectionCancellationToken = null;
        }
    }
-}
+}
@@ -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,17 @@ namespace MQTTnet.Server.Mqtt
        {
            try
            {
+                var sessionItems = (PythonDictionary)context.SessionItems[MqttServerConnectionValidator.WrappedSessionItemsKey];
+
                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);
@@ -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)
            {
@@ -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);
@@ -47,8 +47,6 @@ namespace MQTTnet.Server
            Console.ForegroundColor = ConsoleColor.White;
            Console.WriteLine("The official MQTT server implementation of MQTTnet");
            Console.WriteLine("Copyright (c) 2017-2019 The MQTTnet Team");
-
-            Console.ForegroundColor = ConsoleColor.Blue;
            Console.WriteLine(@"https://github.com/chkr1011/MQTTnet");

            Console.ForegroundColor = ConsoleColor.White;
@@ -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;
@@ -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
            {
@@ -48,7 +48,7 @@ namespace MQTTnet.Implementations
                    throw new ArgumentException("TLS certificate is not set.");
                }

-                var tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate);
+                var 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,6 +60,19 @@ namespace MQTTnet.Implementations
                _logger.Info($"Starting TCP listener for {_localEndPoint} TLS={_tlsCertificate != null}.");

                _socket = new Socket(_addressFamily, SocketType.Stream, ProtocolType.Tcp);
+
+                // 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);

@@ -149,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, 
@@ -160,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" />
@@ -0,0 +1,6 @@
+using System;
+
+public interface IMqttServerCredentials
+{
+    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>
@@ -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, IMqttServerCredentials 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 IMqttServerCredentials 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; }
@@ -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,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()
        {
@@ -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,8 @@ using MQTTnet.Client.Receiving;
 using MQTTnet.Exceptions;
 using MQTTnet.Extensions.Rpc;
 using MQTTnet.Protocol;
+using MQTTnet.Client.Options;
+using MQTTnet.Formatter;

 namespace MQTTnet.Tests
 {
@@ -15,26 +17,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]
@@ -51,5 +66,27 @@ namespace MQTTnet.Tests
                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);
+                var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", qosLevel);
+
+                Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
+            }
+        }
    }
 }
@@ -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 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
Christian Kratky	f66da92299	Merge branch 'develop'	2019-06-27 10:02:59 +02:00
Christian Kratky	979df5f17e	Fix broken nuget builds.	2019-06-27 10:02:47 +02:00
Christian Kratky	d50c8153b2	Merge branch 'master' into develop	2019-06-26 11:42:44 +02:00
Christian Kratky	b3ab7a1069	Set socket options to allow reusing address.	2019-06-24 13:46:51 +02:00
Christian Kratky	6af58ad90d	Fix MQTTnet.Server startup colors.	2019-06-24 13:46:27 +02:00
Christian	5c36a2d402	Merge pull request #671 from SeppPenner/patch-1 Added dependents and MIT license badge to the readme.	2019-06-24 13:40:26 +02:00
HansM	6df70cc5ed	Added dependents and MIT license badge to the readme.	2019-06-24 11:21:24 +02:00
Christian Kratky	d5b0893e90	Merge branch 'master' into develop	2019-06-23 21:50:42 +02:00