Merge branch 'develop'

.gitignore

@@ -292,3 +292,5 @@ __pycache__/
 *.map
 
 /Tests/MQTTnet.TestApp.NetCore/RetainedMessages.json
+
+Build/NuGet/

Build/MQTTnet.nuspec

@@ -11,7 +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>
-* [Server] Moved new socket options to TCP options to avoid incompatibility with Linux hosts.
+* [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> 

Build/build.ps1

@@ -59,12 +59,12 @@ Copy-Item MQTTnet.Extensions.WebSocket4Net.nuspec -Destination MQTTnet.Extension
 (Get-Content MQTTnet.Extensions.WebSocket4Net.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.Extensions.WebSocket4Net.nuspec
 
 New-Item -ItemType Directory -Force -Path .\NuGet
-.\nuget.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-.\nuget.exe pack MQTTnet.NETStandard.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-.\nuget.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-.\nuget.exe pack MQTTnet.Extensions.Rpc.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-.\nuget.exe pack MQTTnet.Extensions.ManagedClient.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-.\nuget.exe pack MQTTnet.Extensions.WebSocket4Net.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.NETStandard.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.Extensions.Rpc.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.Extensions.ManagedClient.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.Extensions.WebSocket4Net.nuspec -Verbosity detailed -Symbols -SymbolPackageFormat snupkg -OutputDir "NuGet" -Version $nugetVersion
 
 Move-Item MQTTnet.AspNetCore.nuspec.old -Destination MQTTnet.AspNetCore.nuspec -Force
 Move-Item MQTTnet.Extensions.Rpc.nuspec.old -Destination MQTTnet.Extensions.Rpc.nuspec -Force

Build/upload.ps1

@@ -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

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -162,11 +162,12 @@ public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, Cance
                 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
             {

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -24,6 +24,8 @@ public class ManagedMqttClient : IManagedMqttClient
 
         private readonly IMqttClient _mqttClient;
         private readonly IMqttNetChildLogger _logger;
+
+        private readonly AsyncLock _messageQueueLock = new AsyncLock();
 
         private CancellationTokenSource _connectionCancellationToken;
         private CancellationTokenSource _publishingCancellationToken;
@@ -147,7 +149,7 @@ public async Task PublishAsync(ManagedMqttApplicationMessage applicationMessage)
 
             try
             {
-                lock (_messageQueue)
+                using (await _messageQueueLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
                 {
                     if (_messageQueue.Count >= Options.MaxPendingMessages)
                     {
@@ -167,6 +169,16 @@ public async Task PublishAsync(ManagedMqttApplicationMessage applicationMessage)
                     }
 
                     _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 @@ public async Task PublishAsync(ManagedMqttApplicationMessage applicationMessage)
                 }
 
             }
-
-            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 @@ private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToke
             }
             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 @@ private async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage me
             {
                 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 @@ private async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage me
                     // 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 @@ private async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage me
                     //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 @@ private void StopMaintainingConnection()
             _connectionCancellationToken = null;
         }
     }
-}
+}

Source/MQTTnet.Server/Controllers/ClientsController.cs

@@ -13,7 +13,7 @@ namespace MQTTnet.Server.Controllers
 {
     [Authorize]
     [ApiController]
-    public class ClientsController : ControllerBase
+    public class ClientsController : Controller
     {
         private readonly MqttServerService _mqttServerService;
 

Source/MQTTnet.Server/Controllers/MessagesController.cs

@@ -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);
+        }
+    }
+}

Source/MQTTnet.Server/Controllers/RetainedApplicationMessagesController.cs

@@ -12,7 +12,7 @@ namespace MQTTnet.Server.Controllers
 {
     [Authorize]
     [ApiController]
-    public class RetainedApplicationMessagesController : ControllerBase
+    public class RetainedApplicationMessagesController : Controller
     {
         private readonly MqttServerService _mqttServerService;
 

Source/MQTTnet.Server/Controllers/ServerController.cs

@@ -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;
+        }
+    }
+}

Source/MQTTnet.Server/Controllers/SessionsController.cs

@@ -13,7 +13,7 @@ namespace MQTTnet.Server.Controllers
 {
     [Authorize]
     [ApiController]
-    public class SessionsController : ControllerBase
+    public class SessionsController : Controller
     {
         private readonly MqttServerService _mqttServerService;
 

Source/MQTTnet.Server/Mqtt/MqttApplicationMessageInterceptor.cs

@@ -22,14 +22,17 @@ public Task InterceptApplicationMessagePublishAsync(MqttApplicationMessageInterc
         {
             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);

Source/MQTTnet.Server/Mqtt/MqttServerConnectionValidator.cs

@@ -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 @@ public Task ValidateConnectionAsync(MqttConnectionValidatorContext context)
         {
             try
             {
+                var sessionItems = new PythonDictionary();
+
                 var pythonContext = new PythonDictionary
                 {
                     { "endpoint", context.Endpoint },
@@ -33,13 +37,16 @@ public Task ValidateConnectionAsync(MqttConnectionValidatorContext context)
                     { "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) }
                 };
 
                 _pythonScriptHostService.InvokeOptionalFunction("on_validate_client_connection", pythonContext);
 
                 context.ReasonCode = PythonConvert.ParseEnum<MqttConnectReasonCode>((string)pythonContext["result"]);
+
+                context.SessionItems[WrappedSessionItemsKey] = sessionItems;
             }
             catch (Exception exception)
             {

Source/MQTTnet.Server/Mqtt/MqttSubscriptionInterceptor.cs

@@ -21,14 +21,16 @@ public Task InterceptSubscriptionAsync(MqttSubscriptionInterceptorContext contex
         {
             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);

Source/MQTTnet/Client/MqttClient.cs

@@ -257,12 +257,12 @@ private async Task DisconnectInternalAsync(Task sender, Exception exception, Mqt
         {
             var clientWasConnected = IsConnected;
 
-            InitiateDisconnect();
-
-            IsConnected = false;
+            TryInitiateDisconnect();
 
             try
             {
+                IsConnected = false;
+
                 if (_adapter != null)
                 {
                     _logger.Verbose("Disconnecting [Timeout={0}]", Options.CommunicationTimeout);
@@ -295,7 +295,7 @@ private async Task DisconnectInternalAsync(Task sender, Exception exception, Mqt
             }
         }
 
-        private void InitiateDisconnect()
+        private void TryInitiateDisconnect()
         {
             lock (_disconnectLock)
             {