From ad77dc0bea6067298b9fb44f0542e0e9d3da7133 Mon Sep 17 00:00:00 2001 From: Jan Eggers Date: Thu, 9 Nov 2017 10:22:44 +0100 Subject: [PATCH 1/4] improved server usage in aspnet so no exceptions are thrown on startup --- MQTTnet.Core/Server/IMqttServer.cs | 1 + MQTTnet.Core/Server/MqttServer.cs | 3 ++ .../Server/MqttServerStartedEventArgs.cs | 8 +++++ Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs | 36 ++++++++++++------- 4 files changed, 36 insertions(+), 12 deletions(-) create mode 100644 MQTTnet.Core/Server/MqttServerStartedEventArgs.cs diff --git a/MQTTnet.Core/Server/IMqttServer.cs b/MQTTnet.Core/Server/IMqttServer.cs index 71af0ee3..13c614fc 100644 --- a/MQTTnet.Core/Server/IMqttServer.cs +++ b/MQTTnet.Core/Server/IMqttServer.cs @@ -8,6 +8,7 @@ namespace MQTTnet.Core.Server { event EventHandler ClientConnected; event EventHandler ClientDisconnected; + event EventHandler Started; IList GetConnectedClients(); void Publish(IEnumerable applicationMessages); diff --git a/MQTTnet.Core/Server/MqttServer.cs b/MQTTnet.Core/Server/MqttServer.cs index 1e05764c..bd06a3a6 100644 --- a/MQTTnet.Core/Server/MqttServer.cs +++ b/MQTTnet.Core/Server/MqttServer.cs @@ -49,6 +49,7 @@ namespace MQTTnet.Core.Server return _clientSessionsManager.GetConnectedClients(); } + public event EventHandler Started; public event EventHandler ClientConnected; public event EventHandler ClientDisconnected; public event EventHandler ApplicationMessageReceived; @@ -96,6 +97,8 @@ namespace MQTTnet.Core.Server } _logger.LogInformation("Started."); + + Started?.Invoke(this, new MqttServerStartedEventArgs()); } public async Task StopAsync() diff --git a/MQTTnet.Core/Server/MqttServerStartedEventArgs.cs b/MQTTnet.Core/Server/MqttServerStartedEventArgs.cs new file mode 100644 index 00000000..8f46bde8 --- /dev/null +++ b/MQTTnet.Core/Server/MqttServerStartedEventArgs.cs @@ -0,0 +1,8 @@ +using System; + +namespace MQTTnet.Core.Server +{ + public class MqttServerStartedEventArgs : EventArgs + { + } +} diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs index d321fd1e..55c57a18 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs @@ -5,7 +5,6 @@ using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Hosting; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.FileProviders; -using Microsoft.Extensions.Logging; using MQTTnet.AspNetCore; using MQTTnet.Core; @@ -21,21 +20,34 @@ namespace MQTTnet.TestApp.AspNetCore2 } // In class _Startup_ of the ASP.NET Core 2.0 project. - public void Configure(IApplicationBuilder app, IHostingEnvironment env, ILoggerFactory loggerFactory) + public void Configure(IApplicationBuilder app, IHostingEnvironment env) { app.UseMqttEndpoint(); - app.UseMqttServer(async server => + app.UseMqttServer(server => { - var msg = new MqttApplicationMessageBuilder() - .WithPayload("Mqtt is awesome") - .WithTopic("message"); - - while (true) + server.Started += async (sender, args) => { - server.PublishAsync(msg.Build()).Wait(); - await Task.Delay(TimeSpan.FromSeconds(2)); - msg.WithPayload("Mqtt is still awesome at " + DateTime.Now); - } + var msg = new MqttApplicationMessageBuilder() + .WithPayload("Mqtt is awesome") + .WithTopic("message"); + + while (true) + { + try + { + await server.PublishAsync(msg.Build()); + msg.WithPayload("Mqtt is still awesome at " + DateTime.Now); + } + catch (Exception e) + { + Console.WriteLine(e); + } + finally + { + await Task.Delay(TimeSpan.FromSeconds(2)); + } + } + }; }); app.Use((context, next) => From 683d97dc14baa4b03f4d89272aaf3bfe3f3d9585 Mon Sep 17 00:00:00 2001 From: Jan Eggers Date: Thu, 9 Nov 2017 10:24:49 +0100 Subject: [PATCH 2/4] added another platform to overcome breaking change in ms.extensions.di 2.0.0 --- Build/MQTTnet.nuspec | 15 ++++++++++++ Build/build.ps1 | 1 + .../Implementations/MqttServerAdapter.cs | 2 +- .../Implementations/MqttTcpChannel.cs | 2 +- .../MQTTnet.Netstandard.csproj | 23 ++++++++++++++++--- .../MQTTnet.Core.Tests.csproj | 2 -- .../MQTTnet.TestApp.NetCore.csproj | 2 -- 7 files changed, 38 insertions(+), 9 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 5fdc919a..3f26a870 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -41,6 +41,17 @@ + + + + + + + + + + + @@ -62,6 +73,10 @@ + + + + diff --git a/Build/build.ps1 b/Build/build.ps1 index e7d86d70..3efa1577 100644 --- a/Build/build.ps1 +++ b/Build/build.ps1 @@ -10,6 +10,7 @@ if ($path) { &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$version /p:AssemblyVersion=$version /verbosity:m &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$version /p:AssemblyVersion=$version /verbosity:m + &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$version /p:AssemblyVersion=$version /verbosity:m &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$version /p:AssemblyVersion=$version /verbosity:m &$msbuild ..\Frameworks\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$version /p:AssemblyVersion=$version /verbosity:m diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs index f894233a..a6e6251e 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs @@ -1,4 +1,4 @@ -#if NET452 || NETSTANDARD1_3 +#if NET452 || NETSTANDARD1_3 || NETSTANDARD2_0 using System; using System.Net; using System.Net.Security; diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs index 0d6180c7..2bf418bc 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs @@ -1,4 +1,4 @@ -#if NET452 || NETSTANDARD1_3 +#if NET452 || NETSTANDARD1_3 || NETSTANDARD2_0 using System; using System.Net.Security; using System.Net.Sockets; diff --git a/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj b/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj index f981d604..266864d1 100644 --- a/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj +++ b/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj @@ -1,7 +1,7 @@  - netstandard1.3;net452;uap10.0 + netstandard1.3;netstandard2.0;net452;uap10.0 MQTTnet MQTTnet 2.5.0.0 @@ -32,23 +32,40 @@ - - + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj b/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj index 802d96d8..88ffbe41 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj +++ b/Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj @@ -7,7 +7,6 @@ - @@ -15,7 +14,6 @@ - \ No newline at end of file diff --git a/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj b/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj index c4a2bac2..80d6a260 100644 --- a/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj +++ b/Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj @@ -7,13 +7,11 @@ - - From b6cd4638b9871af7a9bbff74bac0afd6bc3e747a Mon Sep 17 00:00:00 2001 From: Jan Eggers Date: Thu, 9 Nov 2017 10:40:05 +0100 Subject: [PATCH 3/4] fixed another retain msg test timing issue --- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 39 +++++++++++++-------- 1 file changed, 25 insertions(+), 14 deletions(-) diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index b7098353..fc252678 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -9,6 +9,7 @@ using MQTTnet.Core.Client; using MQTTnet.Core.Protocol; using MQTTnet.Core.Server; using Microsoft.Extensions.DependencyInjection; +using MQTTnet.Core.Internal; using MQTTnet.Core.Packets; namespace MQTTnet.Core.Tests @@ -212,7 +213,6 @@ namespace MQTTnet.Core.Tests .BuildServiceProvider(); var s = new MqttFactory(services).CreateMqttServer(); - var retainMessagemanager = services.GetRequiredService(); var receivedMessagesCount = 0; try @@ -223,19 +223,7 @@ namespace MQTTnet.Core.Tests await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); await c1.DisconnectAsync(); - var subscribe = new MqttSubscribePacket() - { - TopicFilters = new List() - { - new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce) - } - }; - - //make shure the retainedMessageManagerreceived the package - while (!(await retainMessagemanager.GetSubscribedMessagesAsync(subscribe)).Any()) - { - await Task.Delay(TimeSpan.FromMilliseconds(10)); - } + await services.WaitForRetainedMessage("retained").TimeoutAfter(TimeSpan.FromSeconds(5)); var c2 = await serverAdapter.ConnectTestClient(s, "c2"); c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; @@ -314,6 +302,8 @@ namespace MQTTnet.Core.Tests await s.StopAsync(); } + await services.WaitForRetainedMessage("retained").TimeoutAfter(TimeSpan.FromSeconds(5)); + s = new MqttFactory(services).CreateMqttServer(options => options.Storage = storage); var receivedMessagesCount = 0; @@ -436,4 +426,25 @@ namespace MQTTnet.Core.Tests Assert.AreEqual(expectedReceivedMessagesCount, receivedMessagesCount); } } + + public static class TestExtensions + { + public static async Task WaitForRetainedMessage(this IServiceProvider services, string topic) + { + var retainMessagemanager = services.GetRequiredService(); + + var subscribe = new MqttSubscribePacket() + { + TopicFilters = new List() + { + new TopicFilter(topic, MqttQualityOfServiceLevel.AtMostOnce) + } + }; + + while (!(await retainMessagemanager.GetSubscribedMessagesAsync(subscribe)).Any()) + { + await Task.Delay(TimeSpan.FromMilliseconds(10)); + } + } + } } From acbf9153f43b96c956e28bf41a6bf05cfa4da7a4 Mon Sep 17 00:00:00 2001 From: Jan Eggers Date: Thu, 9 Nov 2017 11:06:50 +0100 Subject: [PATCH 4/4] fixed the other timing issue --- MQTTnet.Core/Server/MqttClientSession.cs | 6 ++++-- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 2 ++ 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/MQTTnet.Core/Server/MqttClientSession.cs b/MQTTnet.Core/Server/MqttClientSession.cs index fb859b8f..cbdfc73f 100644 --- a/MQTTnet.Core/Server/MqttClientSession.cs +++ b/MQTTnet.Core/Server/MqttClientSession.cs @@ -94,9 +94,11 @@ namespace MQTTnet.Core.Server } finally { - if (_willMessage != null) + var willMessage = _willMessage; + if (willMessage != null) { - _sessionsManager.DispatchApplicationMessage(this, _willMessage); + _willMessage = null; //clear willmessage so it is send just once + _sessionsManager.DispatchApplicationMessage(this, willMessage); } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index fc252678..7fdaa3d3 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -76,6 +76,8 @@ namespace MQTTnet.Core.Tests await c2.DisconnectAsync(); await Task.Delay(1000); + + await c1.DisconnectAsync(); } finally {