diff --git a/.gitignore b/.gitignore index 39550dc5..d44c4553 100644 --- a/.gitignore +++ b/.gitignore @@ -197,7 +197,6 @@ ClientBin/ *.dbmdl *.dbproj.schemaview *.jfm -*.pfx *.publishsettings orleans.codegen.cs diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec new file mode 100644 index 00000000..eb5f637f --- /dev/null +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -0,0 +1,30 @@ + + + + MQTTnet.AspNetCore + 2.5.0 + Christian Kratky + Christian Kratky + https://github.com/chkr1011/MQTTnet/blob/master/LICENSE + https://github.com/chkr1011/MQTTnet + https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png + false + This is a support library to integrate MQTTnet into AspNetCore. + initial version + + Copyright Christian Kratky 2016-2017 + MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation + + + + + + + + + + + + + + \ No newline at end of file diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 2403cc84..fe5d708b 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -47,12 +47,12 @@ - - + + - - + + \ No newline at end of file diff --git a/Build/build.ps1 b/Build/build.ps1 index 70b616b9..6284473a 100644 --- a/Build/build.ps1 +++ b/Build/build.ps1 @@ -2,10 +2,20 @@ param([string]$version) if ([string]::IsNullOrEmpty($version)) {$version = "0.0.1"} -$msbuild = "MSBuild.exe" -&dotnet build ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj -c="Release" -&$msbuild ..\Frameworks\MQTTnet.UniversalWindows\MQTTnet.UniversalWindows.csproj /t:Build /p:Configuration="Release" +$vswhere = ${Env:\ProgramFiles(x86)} + '\Microsoft Visual Studio\Installer\vswhere' + +$path = &$vswhere -latest -products * -requires Microsoft.Component.MSBuild -property installationPath +if ($path) { + $msbuild = join-path $path 'MSBuild\15.0\Bin\MSBuild.exe' + + &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net451" /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="uap10.0" /p:FileVersion=$version /p:AssemblyVersion=$version /verbosity:m + &dotnet build ..\Frameworks\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj -c="Release" /p:FileVersion=$version /p:AssemblyVersion=$version + + Remove-Item .\NuGet -Force -Recurse + New-Item -ItemType Directory -Force -Path .\NuGet + .\NuGet.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $version + .\NuGet.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $version +} -Remove-Item .\NuGet -Force -Recurse -New-Item -ItemType Directory -Force -Path .\NuGet -.\NuGet.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $version \ No newline at end of file diff --git a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs index a62ecded..06ca74d4 100644 --- a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs @@ -3,7 +3,7 @@ using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.DependencyInjection; using MQTTnet.Core.Server; -namespace MQTTnet.AspnetCore +namespace MQTTnet.AspNetCore { public static class ApplicationBuilderExtensions { diff --git a/Frameworks/MQTTnet.AspnetCore/MqttHostedServer.cs b/Frameworks/MQTTnet.AspnetCore/MqttHostedServer.cs index b5326ab1..db725735 100644 --- a/Frameworks/MQTTnet.AspnetCore/MqttHostedServer.cs +++ b/Frameworks/MQTTnet.AspnetCore/MqttHostedServer.cs @@ -7,7 +7,7 @@ using Microsoft.Extensions.Options; using MQTTnet.Core.Adapter; using MQTTnet.Core.Server; -namespace MQTTnet.AspnetCore +namespace MQTTnet.AspNetCore { public class MqttHostedServer : MqttServer, IHostedService { diff --git a/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs index abb30e37..5cd5d555 100644 --- a/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs +++ b/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs @@ -8,7 +8,7 @@ using MQTTnet.Core.Channel; using MQTTnet.Core.Server; using MQTTnet.Implementations; -namespace MQTTnet.AspnetCore +namespace MQTTnet.AspNetCore { public class MqttWebSocketServerAdapter : IMqttServerAdapter, IDisposable { diff --git a/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs index 1951368c..f915ab43 100644 --- a/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -3,7 +3,7 @@ using Microsoft.Extensions.Hosting; using MQTTnet.Core.Adapter; using MQTTnet.Core.Server; -namespace MQTTnet.AspnetCore +namespace MQTTnet.AspNetCore { public static class ServiceCollectionExtensions { diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs b/Frameworks/MQTTnet.NetStandard/Implementations/NetStandard/MqttServerAdapter.cs similarity index 99% rename from Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs rename to Frameworks/MQTTnet.NetStandard/Implementations/NetStandard/MqttServerAdapter.cs index 9e65715c..1a6e5b40 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/NetStandard/MqttServerAdapter.cs @@ -1,4 +1,6 @@ -using System; +#if NET451 || NETSTANDARD1_3 + +using System; using System.Net; using System.Net.Security; using System.Net.Sockets; @@ -143,4 +145,5 @@ namespace MQTTnet.Implementations } } } -} \ No newline at end of file +} +#endif \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/NetStandard/MqttTcpChannel.cs similarity index 99% rename from Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs rename to Frameworks/MQTTnet.NetStandard/Implementations/NetStandard/MqttTcpChannel.cs index aac8a3c2..62bb8787 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/NetStandard/MqttTcpChannel.cs @@ -1,3 +1,5 @@ +#if NET451 || NETSTANDARD1_3 + using System; using System.Net.Security; using System.Net.Sockets; @@ -155,4 +157,5 @@ namespace MQTTnet.Implementations } } -} \ No newline at end of file +} +#endif \ No newline at end of file diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttServerAdapter.cs b/Frameworks/MQTTnet.NetStandard/Implementations/Uap/MqttServerAdapter.cs similarity index 94% rename from Frameworks/MQTTnet.UniversalWindows/Implementations/MqttServerAdapter.cs rename to Frameworks/MQTTnet.NetStandard/Implementations/Uap/MqttServerAdapter.cs index 373cdac4..7d9d25f1 100644 --- a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttServerAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/Uap/MqttServerAdapter.cs @@ -1,3 +1,5 @@ +#if NET451 || NETSTANDARD1_3 +#else using System; using System.Threading.Tasks; using MQTTnet.Core.Adapter; @@ -30,6 +32,7 @@ namespace MQTTnet.Implementations if (options.DefaultEndpointOptions.IsEnabled) { _defaultEndpointSocket = new StreamSocketListener(); + _defaultEndpointSocket.Control.NoDelay = true; await _defaultEndpointSocket.BindServiceNameAsync(options.GetDefaultEndpointPort().ToString(), SocketProtectionLevel.PlainSocket); _defaultEndpointSocket.ConnectionReceived += AcceptDefaultEndpointConnectionsAsync; } @@ -57,6 +60,8 @@ namespace MQTTnet.Implementations { try { + args.Socket.Control.NoDelay = true; + var clientAdapter = _mqttCommunicationAdapterFactory.CreateServerMqttCommunicationAdapter(new MqttTcpChannel(args.Socket)); ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); } @@ -66,4 +71,5 @@ namespace MQTTnet.Implementations } } } -} \ No newline at end of file +} +#endif \ No newline at end of file diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/Uap/MqttTcpChannel.cs similarity index 99% rename from Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs rename to Frameworks/MQTTnet.NetStandard/Implementations/Uap/MqttTcpChannel.cs index 8fb45ab5..6e311d67 100644 --- a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/Uap/MqttTcpChannel.cs @@ -1,3 +1,5 @@ +#if NET451 || NETSTANDARD1_3 +#else using System; using System.Collections.Generic; using System.IO; @@ -133,4 +135,5 @@ namespace MQTTnet.Implementations return result; } } -} \ No newline at end of file +} +#endif \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj b/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj index f8dbcc9f..37b7e137 100644 --- a/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj +++ b/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj @@ -1,7 +1,7 @@  - netstandard1.3;net451 + netstandard1.3;net451;uap10.0 MQTTnet MQTTnet 2.5.0.0 @@ -14,6 +14,19 @@ + + false + UAP,Version=v10.0 + UAP + 10.0.15063.0 + 10.0.10240.0 + .NETCore + v5.0 + $(DefineConstants);WINDOWS_UWP + $(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets + + + @@ -34,4 +47,8 @@ + + + + \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/MqttFactory.cs b/Frameworks/MQTTnet.NetStandard/MqttFactory.cs index 7d88480f..50cdddea 100644 --- a/Frameworks/MQTTnet.NetStandard/MqttFactory.cs +++ b/Frameworks/MQTTnet.NetStandard/MqttFactory.cs @@ -4,6 +4,7 @@ using MQTTnet.Core.Client; using MQTTnet.Core.Serializer; using Microsoft.Extensions.Logging; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; using MQTTnet.Implementations; using MQTTnet.Core.ManagedClient; using MQTTnet.Core.Server; @@ -90,9 +91,15 @@ namespace MQTTnet }; } - public MqttClientSession CreateClientSession(string sessionId, MqttClientSessionsManager mqttClientSessionsManager) + public MqttClientSession CreateClientSession(string clientId, MqttClientSessionsManager clientSessionsManager) { - return new MqttClientSession(sessionId, mqttClientSessionsManager, _serviceProvider.GetRequiredService>(), _serviceProvider.GetRequiredService>()); + return new MqttClientSession( + clientId, + _serviceProvider.GetRequiredService>(), + clientSessionsManager, + _serviceProvider.GetRequiredService(), + _serviceProvider.GetRequiredService>(), + _serviceProvider.GetRequiredService>()); } public IMqttClient CreateMqttClient() @@ -109,5 +116,15 @@ namespace MQTTnet { return _serviceProvider.GetRequiredService(); } + + public IMqttServer CreateMqttServer(Action configure) + { + if (configure == null) throw new ArgumentNullException(nameof(configure)); + + var options = _serviceProvider.GetRequiredService>(); + configure(options.Value); + + return _serviceProvider.GetRequiredService(); + } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/ServiceCollectionExtensions.cs b/Frameworks/MQTTnet.NetStandard/ServiceCollectionExtensions.cs index fdecb32a..aa53e1b8 100644 --- a/Frameworks/MQTTnet.NetStandard/ServiceCollectionExtensions.cs +++ b/Frameworks/MQTTnet.NetStandard/ServiceCollectionExtensions.cs @@ -34,6 +34,7 @@ namespace MQTTnet services.AddTransient(); services.AddTransient(); + services.AddTransient(); services.AddTransient(); return services; } diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs deleted file mode 100644 index 6145802b..00000000 --- a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs +++ /dev/null @@ -1,92 +0,0 @@ -using MQTTnet.Core.Channel; -using MQTTnet.Core.Client; -using System; -using System.IO; -using System.Net.WebSockets; -using System.Security.Cryptography.X509Certificates; -using System.Threading; -using System.Threading.Tasks; - -namespace MQTTnet.Implementations -{ - public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable - { - private readonly MqttClientWebSocketOptions _options; - private ClientWebSocket _webSocket = new ClientWebSocket(); - - public MqttWebSocketChannel(MqttClientWebSocketOptions options) - { - _options = options ?? throw new ArgumentNullException(nameof(options)); - } - - public Stream RawReceiveStream { get; private set; } - public Stream SendStream => RawReceiveStream; - public Stream ReceiveStream => RawReceiveStream; - - public async Task ConnectAsync() - { - var uri = _options.Uri; - if (!uri.StartsWith("ws://", StringComparison.OrdinalIgnoreCase)) - { - uri = "ws://" + uri; - } - - _webSocket = new ClientWebSocket(); - - if (_options.RequestHeaders != null) - { - foreach (var requestHeader in _options.RequestHeaders) - { - _webSocket.Options.SetRequestHeader(requestHeader.Key, requestHeader.Value); - } - } - - if (_options.SubProtocols != null) - { - foreach (var subProtocol in _options.SubProtocols) - { - _webSocket.Options.AddSubProtocol(subProtocol); - } - } - - if (_options.CookieContainer != null) - { - _webSocket.Options.Cookies = _options.CookieContainer; - } - - if (_options.TlsOptions?.UseTls == true && _options.TlsOptions?.Certificates != null) - { - _webSocket.Options.ClientCertificates = new X509CertificateCollection(); - foreach (var certificate in _options.TlsOptions.Certificates) - { - _webSocket.Options.ClientCertificates.Add(new X509Certificate(certificate)); - } - } - - await _webSocket.ConnectAsync(new Uri(uri), CancellationToken.None).ConfigureAwait(false); - - RawReceiveStream = new WebSocketStream(_webSocket); - } - - public async Task DisconnectAsync() - { - RawReceiveStream = null; - - if (_webSocket == null) - { - return; - } - - if (_webSocket.State == WebSocketState.Open || _webSocket.State == WebSocketState.Connecting) - { - await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false); - } - } - - public void Dispose() - { - _webSocket?.Dispose(); - _webSocket = null; - } - } -} \ No newline at end of file diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/WebSocketStream.cs b/Frameworks/MQTTnet.UniversalWindows/Implementations/WebSocketStream.cs deleted file mode 100644 index 718b5eec..00000000 --- a/Frameworks/MQTTnet.UniversalWindows/Implementations/WebSocketStream.cs +++ /dev/null @@ -1,84 +0,0 @@ -using System; -using System.IO; -using System.Net.WebSockets; -using System.Threading; -using System.Threading.Tasks; -using MQTTnet.Core.Exceptions; - -namespace MQTTnet.Implementations -{ - public class WebSocketStream : Stream - { - private readonly WebSocket _webSocket; - - public WebSocketStream(WebSocket webSocket) - { - _webSocket = webSocket ?? throw new ArgumentNullException(nameof(webSocket)); - } - - public override void Flush() - { - } - - public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) - { - var currentOffset = offset; - var targetOffset = offset + count; - while (_webSocket.State == WebSocketState.Open && currentOffset < targetOffset) - { - var response = await _webSocket.ReceiveAsync(new ArraySegment(buffer, currentOffset, count), cancellationToken).ConfigureAwait(false); - currentOffset += response.Count; - count -= response.Count; - - if (response.MessageType == WebSocketMessageType.Close) - { - await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false); - } - } - - if (_webSocket.State == WebSocketState.Closed) - { - throw new MqttCommunicationException("connection closed"); - } - - return currentOffset - offset; - } - - public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) - { - return _webSocket.SendAsync(new ArraySegment(buffer, offset, count), WebSocketMessageType.Binary, true, cancellationToken); - } - - public override int Read(byte[] buffer, int offset, int count) - { - return ReadAsync(buffer, offset, count).GetAwaiter().GetResult(); - } - - public override void Write(byte[] buffer, int offset, int count) - { - WriteAsync(buffer, offset, count).GetAwaiter().GetResult(); - } - - public override bool CanRead => true; - public override bool CanSeek => false; - public override bool CanWrite => true; - - public override long Length => throw new NotSupportedException(); - - public override long Position - { - get => throw new NotSupportedException(); - set => throw new NotSupportedException(); - } - - public override long Seek(long offset, SeekOrigin origin) - { - throw new NotSupportedException(); - } - - public override void SetLength(long value) - { - throw new NotSupportedException(); - } - } -} diff --git a/Frameworks/MQTTnet.UniversalWindows/MQTTnet.UniversalWindows.csproj b/Frameworks/MQTTnet.UniversalWindows/MQTTnet.UniversalWindows.csproj deleted file mode 100644 index d8051f81..00000000 --- a/Frameworks/MQTTnet.UniversalWindows/MQTTnet.UniversalWindows.csproj +++ /dev/null @@ -1,200 +0,0 @@ - - - - - Debug - AnyCPU - {BD60C727-D8E8-40C3-B8E3-C95A864AE611} - Library - Properties - MQTTnet - MQTTnet - en-US - UAP - 10.0.15063.0 - 10.0.10240.0 - 14 - 512 - {A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC} - win10-arm;win10-arm-aot;win10-x86;win10-x86-aot;win10-x64;win10-x64-aot - - - AnyCPU - true - full - false - bin\Debug\ - DEBUG;TRACE;NETFX_CORE;WINDOWS_UWP - prompt - 4 - CS1591 - - - AnyCPU - pdbonly - true - bin\Release\ - TRACE;NETFX_CORE;WINDOWS_UWP - prompt - 4 - CS1591 - - - x86 - true - bin\x86\Debug\ - DEBUG;TRACE;NETFX_CORE;WINDOWS_UWP - ;2008,CS1591 - full - x86 - false - prompt - - - x86 - bin\x86\Release\ - TRACE;NETFX_CORE;WINDOWS_UWP - true - ;2008,CS1591 - pdbonly - x86 - false - prompt - bin\x86\Release\MQTTnet.XML - - - ARM - true - bin\ARM\Debug\ - DEBUG;TRACE;NETFX_CORE;WINDOWS_UWP - ;2008,CS1591 - full - ARM - false - prompt - - - ARM - bin\ARM\Release\ - TRACE;NETFX_CORE;WINDOWS_UWP - true - ;2008,CS1591 - pdbonly - ARM - false - prompt - bin\ARM\Release\MQTTnet.XML - - - x64 - true - bin\x64\Debug\ - DEBUG;TRACE;NETFX_CORE;WINDOWS_UWP - ;2008,CS1591 - full - x64 - false - prompt - - - x64 - bin\x64\Release\ - TRACE;NETFX_CORE;WINDOWS_UWP - true - ;2008,CS1591 - pdbonly - x64 - false - prompt - false - bin\x64\Release\MQTTnet.XML - - - - - - - - - - - - - - {2ecb99e4-72d0-4c23-99ba-93d511d3967d} - MQTTnet.Core - - - - - 1.1.1 - - - 1.1.2 - - - 5.4.0 - - - - 14.0 - - - bin\Any CPU\ - TRACE;NETFX_CORE;WINDOWS_UWP - true - true - pdbonly - AnyCPU - false - prompt - MinimumRecommendedRules.ruleset - - - bin\x86\Any CPU\ - TRACE;NETFX_CORE;WINDOWS_UWP - bin\x86\Release\MQTTnet.XML - true - ;2008 - true - pdbonly - x86 - false - prompt - MinimumRecommendedRules.ruleset - - - bin\ARM\Any CPU\ - TRACE;NETFX_CORE;WINDOWS_UWP - bin\ARM\Release\MQTTnet.XML - true - ;2008 - true - pdbonly - ARM - false - prompt - MinimumRecommendedRules.ruleset - - - bin\x64\Any CPU\ - TRACE;NETFX_CORE;WINDOWS_UWP - bin\x64\Release\MQTTnet.XML - true - ;2008 - true - pdbonly - x64 - false - prompt - MinimumRecommendedRules.ruleset - - - - \ No newline at end of file diff --git a/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs b/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs deleted file mode 100644 index 24826cb7..00000000 --- a/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs +++ /dev/null @@ -1,108 +0,0 @@ -using System; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; -using MQTTnet.Core.Adapter; -using MQTTnet.Core.Channel; -using MQTTnet.Core.Client; -using MQTTnet.Core.ManagedClient; -using MQTTnet.Core.Serializer; -using MQTTnet.Core.Server; -using MQTTnet.Implementations; - -namespace MQTTnet -{ - public class MqttFactory : IMqttCommunicationAdapterFactory, IMqttClientSesssionFactory, IMqttClientFactory, IMqttServerFactory - { - private readonly IServiceProvider _serviceProvider; - - private static IServiceProvider BuildServiceProvider() - { - var serviceProvider = new ServiceCollection() - .AddMqttClient() - .AddMqttServer() - .AddLogging() - .BuildServiceProvider(); - - serviceProvider.GetRequiredService() - .AddMqttTrace(); - - return serviceProvider; - } - - public MqttFactory() - : this(BuildServiceProvider()) - { - } - - public MqttFactory(IServiceProvider serviceProvider) - { - _serviceProvider = serviceProvider; - } - - public IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options) - { - var logger = _serviceProvider.GetRequiredService>(); - return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options.ChannelOptions), CreateSerializer(options.ProtocolVersion), logger); - } - - public IMqttCommunicationAdapter CreateServerMqttCommunicationAdapter(IMqttCommunicationChannel channel) - { - var serializer = _serviceProvider.GetRequiredService(); - var logger = _serviceProvider.GetRequiredService>(); - return new MqttChannelCommunicationAdapter(channel, serializer, logger); - } - - public IMqttCommunicationChannel CreateMqttCommunicationChannel(IMqttClientChannelOptions options) - { - if (options == null) throw new ArgumentNullException(nameof(options)); - - switch (options) - { - case MqttClientTcpOptions tcpOptions: - return CreateTcpChannel(tcpOptions); - case MqttClientWebSocketOptions webSocketOptions: - return CreateWebSocketChannel(webSocketOptions); - default: - throw new NotSupportedException(); - } - } - - public MqttTcpChannel CreateTcpChannel(MqttClientTcpOptions tcpOptions) - { - return new MqttTcpChannel(tcpOptions); - } - - public MqttWebSocketChannel CreateWebSocketChannel(MqttClientWebSocketOptions webSocketOptions) - { - return new MqttWebSocketChannel(webSocketOptions); - } - - public MqttPacketSerializer CreateSerializer(MqttProtocolVersion protocolVersion) - { - return new MqttPacketSerializer - { - ProtocolVersion = protocolVersion - }; - } - - public MqttClientSession CreateClientSession(string sessionId, MqttClientSessionsManager mqttClientSessionsManager) - { - return new MqttClientSession(sessionId, mqttClientSessionsManager, _serviceProvider.GetRequiredService>(), _serviceProvider.GetRequiredService>()); - } - - public IMqttClient CreateMqttClient() - { - return _serviceProvider.GetRequiredService(); - } - - public IManagedMqttClient CreateManagedMqttClient() - { - return _serviceProvider.GetRequiredService(); - } - - public IMqttServer CreateMqttServer() - { - return _serviceProvider.GetRequiredService(); - } - } -} \ No newline at end of file diff --git a/Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs b/Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs deleted file mode 100644 index 57632880..00000000 --- a/Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs +++ /dev/null @@ -1,14 +0,0 @@ -using System.Reflection; -using System.Runtime.InteropServices; - -[assembly: AssemblyTitle("MQTTnet")] -[assembly: AssemblyDescription("MQTTnet for Universal Windows")] -[assembly: AssemblyConfiguration("")] -[assembly: AssemblyCompany("Christian Kratky")] -[assembly: AssemblyProduct("MQTTnet")] -[assembly: AssemblyCopyright("Copyright © Christian Kratky 2016-2017")] -[assembly: AssemblyTrademark("")] -[assembly: AssemblyCulture("")] -[assembly: ComVisible(false)] -[assembly: AssemblyVersion("2.5.0.0")] -[assembly: AssemblyFileVersion("2.5.0.0")] \ No newline at end of file diff --git a/Frameworks/MQTTnet.UniversalWindows/Properties/MQTTnet.Universal.rd.xml b/Frameworks/MQTTnet.UniversalWindows/Properties/MQTTnet.Universal.rd.xml deleted file mode 100644 index e6d24943..00000000 --- a/Frameworks/MQTTnet.UniversalWindows/Properties/MQTTnet.Universal.rd.xml +++ /dev/null @@ -1,5 +0,0 @@ - - - - - diff --git a/Frameworks/MQTTnet.UniversalWindows/ServiceCollectionExtensions.cs b/Frameworks/MQTTnet.UniversalWindows/ServiceCollectionExtensions.cs deleted file mode 100644 index 9b8e6f7b..00000000 --- a/Frameworks/MQTTnet.UniversalWindows/ServiceCollectionExtensions.cs +++ /dev/null @@ -1,63 +0,0 @@ -using Microsoft.Extensions.DependencyInjection; -using MQTTnet.Core.Adapter; -using MQTTnet.Core.Client; -using MQTTnet.Core.ManagedClient; -using MQTTnet.Core.Serializer; -using MQTTnet.Core.Server; -using MQTTnet.Implementations; -using System; -using Microsoft.Extensions.Logging; -using MQTTnet.Core.Diagnostics; - -namespace MQTTnet -{ - public static class ServiceCollectionExtensions - { - public static IServiceCollection AddMqttServer(this IServiceCollection services) - { - services.AddOptions(); - services.AddSingleton(); - services.AddSingleton(); - services.AddSingleton(); - - - services.AddSingleton(); - services.AddSingleton(); - - services.AddTransient(); - services.AddTransient(); - - services.AddTransient(); - services.AddTransient(); - return services; - } - - public static IServiceCollection AddMqttServer(this IServiceCollection services, Action configureOptions) - { - return services - .AddMqttServer() - .Configure(configureOptions); - } - - public static IServiceCollection AddMqttClient(this IServiceCollection services) - { - services.AddSingleton(); - services.AddSingleton(); - - services.AddTransient(); - services.AddTransient(); - services.AddTransient(); - services.AddTransient(); - services.AddTransient(); - services.AddTransient(); - - return services; - } - - public static ILoggerFactory AddMqttTrace(this ILoggerFactory factory) - { - factory.AddProvider(new MqttNetTrace()); - return factory; - } - } -} diff --git a/MQTTnet.Core/ApplicationMessagePublisherExtensions.cs b/MQTTnet.Core/ApplicationMessagePublisherExtensions.cs new file mode 100644 index 00000000..6893a1b6 --- /dev/null +++ b/MQTTnet.Core/ApplicationMessagePublisherExtensions.cs @@ -0,0 +1,16 @@ +using System; +using System.Threading.Tasks; + +namespace MQTTnet.Core +{ + public static class ApplicationMessagePublisherExtensions + { + public static Task PublishAsync(this IApplicationMessagePublisher client, params MqttApplicationMessage[] applicationMessages) + { + if (client == null) throw new ArgumentNullException(nameof(client)); + if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); + + return client.PublishAsync(applicationMessages); + } + } +} diff --git a/MQTTnet.Core/Client/IMqttClientOptions.cs b/MQTTnet.Core/Client/IMqttClientOptions.cs index d2bcf9d3..ae771922 100644 --- a/MQTTnet.Core/Client/IMqttClientOptions.cs +++ b/MQTTnet.Core/Client/IMqttClientOptions.cs @@ -6,6 +6,11 @@ namespace MQTTnet.Core.Client public interface IMqttClientOptions { string ClientId { get; } + + /// + /// The LogId is used to create a scope to correlate logging. If no value is provided the ClientId is used instead + /// + string LogId { get; } IMqttClientCredentials Credentials { get; } bool CleanSession { get; } MqttApplicationMessage WillMessage { get; } diff --git a/MQTTnet.Core/Client/MqttClient.cs b/MQTTnet.Core/Client/MqttClient.cs index b54288c7..d16575b4 100644 --- a/MQTTnet.Core/Client/MqttClient.cs +++ b/MQTTnet.Core/Client/MqttClient.cs @@ -55,7 +55,7 @@ namespace MQTTnet.Core.Client _adapter = _communicationAdapterFactory.CreateClientMqttCommunicationAdapter(options); - _scopeHandle = _logger.BeginScope(options.ClientId); + _scopeHandle = _logger.BeginScope(options.LogId ?? options.ClientId); _logger.LogTrace("Trying to connect with server."); await _adapter.ConnectAsync(_options.CommunicationTimeout).ConfigureAwait(false); _logger.LogTrace("Connection with server established."); @@ -149,36 +149,36 @@ namespace MQTTnet.Core.Client switch (qosGroup.Key) { case MqttQualityOfServiceLevel.AtMostOnce: - { - // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] - await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosPackets); - break; - } + { + // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] + await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosPackets); + break; + } case MqttQualityOfServiceLevel.AtLeastOnce: + { + foreach (var publishPacket in qosPackets) { - foreach (var publishPacket in qosPackets) - { - publishPacket.PacketIdentifier = GetNewPacketIdentifier(); - await SendAndReceiveAsync(publishPacket); - } - - break; + publishPacket.PacketIdentifier = GetNewPacketIdentifier(); + await SendAndReceiveAsync(publishPacket); } + + break; + } case MqttQualityOfServiceLevel.ExactlyOnce: + { + foreach (var publishPacket in qosPackets) { - foreach (var publishPacket in qosPackets) - { - publishPacket.PacketIdentifier = GetNewPacketIdentifier(); - var pubRecPacket = await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); - await SendAndReceiveAsync(pubRecPacket.CreateResponse()).ConfigureAwait(false); - } + publishPacket.PacketIdentifier = GetNewPacketIdentifier(); + var pubRecPacket = await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); + await SendAndReceiveAsync(pubRecPacket.CreateResponse()).ConfigureAwait(false); + } - break; - } + break; + } default: - { - throw new InvalidOperationException(); - } + { + throw new InvalidOperationException(); + } } } } @@ -191,7 +191,7 @@ namespace MQTTnet.Core.Client Username = _options.Credentials?.Username, Password = _options.Credentials?.Password, CleanSession = _options.CleanSession, - KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds, + KeepAlivePeriod = (ushort) _options.KeepAlivePeriod.TotalSeconds, WillMessage = willApplicationMessage }; @@ -324,7 +324,7 @@ namespace MQTTnet.Core.Client if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) { FireApplicationMessageReceivedEvent(publishPacket); - await SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }); + await SendAsync(new MqttPubAckPacket {PacketIdentifier = publishPacket.PacketIdentifier}); return; } @@ -337,7 +337,7 @@ namespace MQTTnet.Core.Client } FireApplicationMessageReceivedEvent(publishPacket); - await SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }); + await SendAsync(new MqttPubRecPacket {PacketIdentifier = publishPacket.PacketIdentifier}); return; } @@ -363,12 +363,12 @@ namespace MQTTnet.Core.Client { var packetAwaiter = _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.CommunicationTimeout); await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false); - return (TResponsePacket)await packetAwaiter.ConfigureAwait(false); + return (TResponsePacket) await packetAwaiter.ConfigureAwait(false); } private ushort GetNewPacketIdentifier() { - return (ushort)Interlocked.Increment(ref _latestPacketIdentifier); + return (ushort) Interlocked.Increment(ref _latestPacketIdentifier); } private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken) @@ -465,7 +465,8 @@ namespace MQTTnet.Core.Client private void StartSendKeepAliveMessages(CancellationToken cancellationToken) { #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed - Task.Factory.StartNew(() => SendKeepAliveMessagesAsync(cancellationToken), cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default).ConfigureAwait(false); + Task.Factory.StartNew(() => SendKeepAliveMessagesAsync(cancellationToken), cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default) + .ConfigureAwait(false); #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed } } diff --git a/MQTTnet.Core/Client/MqttClientExtensions.cs b/MQTTnet.Core/Client/MqttClientExtensions.cs index ff61548b..567184d0 100644 --- a/MQTTnet.Core/Client/MqttClientExtensions.cs +++ b/MQTTnet.Core/Client/MqttClientExtensions.cs @@ -7,14 +7,6 @@ namespace MQTTnet.Core.Client { public static class MqttClientExtensions { - public static Task PublishAsync(this IApplicationMessagePublisher client, params MqttApplicationMessage[] applicationMessages) - { - if (client == null) throw new ArgumentNullException(nameof(client)); - if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); - - return client.PublishAsync(applicationMessages); - } - public static Task> SubscribeAsync(this IMqttClient client, params TopicFilter[] topicFilters) { if (client == null) throw new ArgumentNullException(nameof(client)); diff --git a/MQTTnet.Core/Client/MqttClientOptions.cs b/MQTTnet.Core/Client/MqttClientOptions.cs index 04cf8bdb..1b1c1bb8 100644 --- a/MQTTnet.Core/Client/MqttClientOptions.cs +++ b/MQTTnet.Core/Client/MqttClientOptions.cs @@ -9,6 +9,9 @@ namespace MQTTnet.Core.Client public string ClientId { get; set; } = Guid.NewGuid().ToString("N"); + /// + public string LogId { get; set; } + public bool CleanSession { get; set; } = true; public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials(); diff --git a/MQTTnet.Core/Client/IApplicationMessagePublisher.cs b/MQTTnet.Core/IApplicationMessagePublisher.cs similarity index 87% rename from MQTTnet.Core/Client/IApplicationMessagePublisher.cs rename to MQTTnet.Core/IApplicationMessagePublisher.cs index c77fd3f6..2f6715b8 100644 --- a/MQTTnet.Core/Client/IApplicationMessagePublisher.cs +++ b/MQTTnet.Core/IApplicationMessagePublisher.cs @@ -1,7 +1,7 @@ using System.Collections.Generic; using System.Threading.Tasks; -namespace MQTTnet.Core.Client +namespace MQTTnet.Core { public interface IApplicationMessagePublisher { diff --git a/MQTTnet.Core/Client/IApplicationMessageReceiver.cs b/MQTTnet.Core/IApplicationMessageReceiver.cs similarity index 78% rename from MQTTnet.Core/Client/IApplicationMessageReceiver.cs rename to MQTTnet.Core/IApplicationMessageReceiver.cs index 2d53cb8a..379ba69c 100644 --- a/MQTTnet.Core/Client/IApplicationMessageReceiver.cs +++ b/MQTTnet.Core/IApplicationMessageReceiver.cs @@ -1,6 +1,7 @@ using System; +using MQTTnet.Core.Client; -namespace MQTTnet.Core.Client +namespace MQTTnet.Core { public interface IApplicationMessageReceiver { diff --git a/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs b/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs index 61c0067a..4eb40b23 100644 --- a/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs +++ b/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs @@ -91,16 +91,15 @@ namespace MQTTnet.Core.ManagedClient return Task.FromResult(0); } - public Task PublishAsync(IEnumerable applicationMessages) + public async Task PublishAsync(IEnumerable applicationMessages) { if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); foreach (var applicationMessage in applicationMessages) { + await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false); _messageQueue.Add(applicationMessage); } - - return Task.FromResult(0); } public Task SubscribeAsync(IEnumerable topicFilters) @@ -208,6 +207,7 @@ namespace MQTTnet.Core.ManagedClient } await TryPublishQueuedMessageAsync(message).ConfigureAwait(false); + await _storageManager.RemoveAsync(message).ConfigureAwait(false); } } catch (OperationCanceledException) diff --git a/MQTTnet.Core/ManagedClient/ManagedMqttClientExtensions.cs b/MQTTnet.Core/ManagedClient/ManagedMqttClientExtensions.cs index 62aa166f..a9984c4a 100644 --- a/MQTTnet.Core/ManagedClient/ManagedMqttClientExtensions.cs +++ b/MQTTnet.Core/ManagedClient/ManagedMqttClientExtensions.cs @@ -1,26 +1,18 @@ using System; using System.Threading.Tasks; -using MQTTnet.Core.Packets; namespace MQTTnet.Core.ManagedClient { public static class ManagedMqttClientExtensions { - public static Task EnqueueAsync(this ManagedMqttClient managedClient, params MqttApplicationMessage[] applicationMessages) - { - if (managedClient == null) throw new ArgumentNullException(nameof(managedClient)); - - return managedClient.EnqueueAsync(applicationMessages); - } - - public static Task SubscribeAsync(this ManagedMqttClient managedClient, params TopicFilter[] topicFilters) + public static Task SubscribeAsync(this IManagedMqttClient managedClient, params TopicFilter[] topicFilters) { if (managedClient == null) throw new ArgumentNullException(nameof(managedClient)); return managedClient.SubscribeAsync(topicFilters); } - public static Task UnsubscribeAsync(this ManagedMqttClient managedClient, params TopicFilter[] topicFilters) + public static Task UnsubscribeAsync(this IManagedMqttClient managedClient, params TopicFilter[] topicFilters) { if (managedClient == null) throw new ArgumentNullException(nameof(managedClient)); diff --git a/MQTTnet.Core/Client/MqttApplicationMessageReceivedEventArgs.cs b/MQTTnet.Core/MqttApplicationMessageReceivedEventArgs.cs similarity index 94% rename from MQTTnet.Core/Client/MqttApplicationMessageReceivedEventArgs.cs rename to MQTTnet.Core/MqttApplicationMessageReceivedEventArgs.cs index 98ebce5a..3de0db54 100644 --- a/MQTTnet.Core/Client/MqttApplicationMessageReceivedEventArgs.cs +++ b/MQTTnet.Core/MqttApplicationMessageReceivedEventArgs.cs @@ -1,6 +1,6 @@ using System; -namespace MQTTnet.Core.Client +namespace MQTTnet.Core { public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs { diff --git a/MQTTnet.Core/Server/IMqttServer.cs b/MQTTnet.Core/Server/IMqttServer.cs index 58c4b74c..7d76c23c 100644 --- a/MQTTnet.Core/Server/IMqttServer.cs +++ b/MQTTnet.Core/Server/IMqttServer.cs @@ -11,6 +11,7 @@ namespace MQTTnet.Core.Server event EventHandler ClientDisconnected; IList GetConnectedClients(); + void Publish(IEnumerable applicationMessages); Task StartAsync(); Task StopAsync(); diff --git a/MQTTnet.Core/Server/IMqttServerFactory.cs b/MQTTnet.Core/Server/IMqttServerFactory.cs index 61d022f3..500614b0 100644 --- a/MQTTnet.Core/Server/IMqttServerFactory.cs +++ b/MQTTnet.Core/Server/IMqttServerFactory.cs @@ -1,7 +1,11 @@ -namespace MQTTnet.Core.Server +using System; + +namespace MQTTnet.Core.Server { public interface IMqttServerFactory { IMqttServer CreateMqttServer(); + + IMqttServer CreateMqttServer(Action configure); } } \ No newline at end of file diff --git a/MQTTnet.Core/Server/MqttApplicationMessageInterceptorContext.cs b/MQTTnet.Core/Server/MqttApplicationMessageInterceptorContext.cs new file mode 100644 index 00000000..218a02dc --- /dev/null +++ b/MQTTnet.Core/Server/MqttApplicationMessageInterceptorContext.cs @@ -0,0 +1,7 @@ +namespace MQTTnet.Core.Server +{ + public class MqttApplicationMessageInterceptorContext + { + public MqttApplicationMessage ApplicationMessage { get; set; } + } +} diff --git a/MQTTnet.Core/Server/MqttClientSession.cs b/MQTTnet.Core/Server/MqttClientSession.cs index 2bd3617a..903b3306 100644 --- a/MQTTnet.Core/Server/MqttClientSession.cs +++ b/MQTTnet.Core/Server/MqttClientSession.cs @@ -9,6 +9,7 @@ using MQTTnet.Core.Packets; using MQTTnet.Core.Protocol; using MQTTnet.Core.Serializer; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; namespace MQTTnet.Core.Server { @@ -16,8 +17,8 @@ namespace MQTTnet.Core.Server { private readonly HashSet _unacknowledgedPublishPackets = new HashSet(); - private readonly MqttClientSubscriptionsManager _subscriptionsManager = new MqttClientSubscriptionsManager(); - private readonly MqttClientSessionsManager _mqttClientSessionsManager; + private readonly MqttClientSubscriptionsManager _subscriptionsManager; + private readonly MqttClientSessionsManager _sessionsManager; private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue; private readonly MqttServerOptions _options; private readonly ILogger _logger; @@ -26,14 +27,22 @@ namespace MQTTnet.Core.Server private CancellationTokenSource _cancellationTokenSource; private MqttApplicationMessage _willMessage; - public MqttClientSession(string clientId, MqttClientSessionsManager mqttClientSessionsManager, ILogger logger, ILogger msgQueueLogger) + public MqttClientSession( + string clientId, + IOptions options, + MqttClientSessionsManager sessionsManager, + MqttClientSubscriptionsManager subscriptionsManager, + ILogger logger, + ILogger messageQueueLogger) { - _mqttClientSessionsManager = mqttClientSessionsManager ?? throw new ArgumentNullException(nameof(mqttClientSessionsManager)); + _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager)); + _subscriptionsManager = subscriptionsManager ?? throw new ArgumentNullException(nameof(sessionsManager)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + ClientId = clientId; - _options = mqttClientSessionsManager.Options; - _pendingMessagesQueue = new MqttClientPendingMessagesQueue(mqttClientSessionsManager.Options, this, msgQueueLogger); + _options = options.Value; + _pendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, messageQueueLogger); } public string ClientId { get; } @@ -84,7 +93,7 @@ namespace MQTTnet.Core.Server { if (_willMessage != null) { - _mqttClientSessionsManager.DispatchApplicationMessage(this, _willMessage); + _sessionsManager.DispatchApplicationMessage(this, _willMessage); } } } @@ -115,7 +124,7 @@ namespace MQTTnet.Core.Server while (!cancellationToken.IsCancellationRequested) { var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false); - await ProcessReceivedPacketAsync(adapter, packet).ConfigureAwait(false); + await ProcessReceivedPacketAsync(adapter, packet, cancellationToken).ConfigureAwait(false); } } catch (OperationCanceledException) @@ -133,79 +142,107 @@ namespace MQTTnet.Core.Server } } - private async Task ProcessReceivedPacketAsync(IMqttCommunicationAdapter adapter, MqttBasePacket packet) + private Task ProcessReceivedPacketAsync(IMqttCommunicationAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken) { - if (packet is MqttSubscribePacket subscribePacket) + if (packet is MqttPingReqPacket) { - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, _subscriptionsManager.Subscribe(subscribePacket)); - EnqueueRetainedMessages(subscribePacket); + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPingRespPacket()); } - else if (packet is MqttUnsubscribePacket unsubscribePacket) + + if (packet is MqttPublishPacket publishPacket) { - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, _subscriptionsManager.Unsubscribe(unsubscribePacket)); + return HandleIncomingPublishPacketAsync(adapter, publishPacket, cancellationToken); } - else if (packet is MqttPublishPacket publishPacket) + + if (packet is MqttPubRelPacket pubRelPacket) { - await HandleIncomingPublishPacketAsync(adapter, publishPacket); + return HandleIncomingPubRelPacketAsync(adapter, pubRelPacket, cancellationToken); } - else if (packet is MqttPubRelPacket pubRelPacket) + + if (packet is MqttPubRecPacket pubRecPacket) { - await HandleIncomingPubRelPacketAsync(adapter, pubRelPacket); + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, pubRecPacket.CreateResponse()); } - else if (packet is MqttPubRecPacket pubRecPacket) - { - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, pubRecPacket.CreateResponse()); - } - else if (packet is MqttPubAckPacket || packet is MqttPubCompPacket) + + if (packet is MqttPubAckPacket || packet is MqttPubCompPacket) { // Discard message. + return Task.FromResult(0); } - else if (packet is MqttPingReqPacket) + + if (packet is MqttSubscribePacket subscribePacket) { - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPingRespPacket()); + return HandleIncomingSubscribePacketAsync(adapter, subscribePacket, cancellationToken); } - else if (packet is MqttDisconnectPacket || packet is MqttConnectPacket) + + if (packet is MqttUnsubscribePacket unsubscribePacket) + { + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, _subscriptionsManager.Unsubscribe(unsubscribePacket)); + } + + if (packet is MqttDisconnectPacket || packet is MqttConnectPacket) { Stop(); + return Task.FromResult(0); } - else + + _logger.LogWarning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); + Stop(); + + return Task.FromResult(0); + } + + private async Task HandleIncomingSubscribePacketAsync(IMqttCommunicationAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) + { + var subscribeResult = _subscriptionsManager.Subscribe(subscribePacket, ClientId); + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, subscribeResult.ResponsePacket); + EnqueueRetainedMessages(subscribePacket); + + if (subscribeResult.CloseConnection) { - _logger.LogWarning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()); Stop(); } } private void EnqueueRetainedMessages(MqttSubscribePacket subscribePacket) { - var retainedMessages = _mqttClientSessionsManager.RetainedMessagesManager.GetMessages(subscribePacket); + var retainedMessages = _sessionsManager.RetainedMessagesManager.GetMessages(subscribePacket); foreach (var publishPacket in retainedMessages) { EnqueuePublishPacket(publishPacket.ToPublishPacket()); } } - private async Task HandleIncomingPublishPacketAsync(IMqttCommunicationAdapter adapter, MqttPublishPacket publishPacket) + private async Task HandleIncomingPublishPacketAsync(IMqttCommunicationAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken) { var applicationMessage = publishPacket.ToApplicationMessage(); - _options.ApplicationMessageInterceptor?.Invoke(applicationMessage); + + var interceptorContext = new MqttApplicationMessageInterceptorContext + { + ApplicationMessage = applicationMessage + }; + + _options.ApplicationMessageInterceptor?.Invoke(interceptorContext); + applicationMessage = interceptorContext.ApplicationMessage; if (applicationMessage.Retain) { - await _mqttClientSessionsManager.RetainedMessagesManager.HandleMessageAsync(ClientId, applicationMessage); + await _sessionsManager.RetainedMessagesManager.HandleMessageAsync(ClientId, applicationMessage); } switch (applicationMessage.QualityOfServiceLevel) { case MqttQualityOfServiceLevel.AtMostOnce: { - _mqttClientSessionsManager.DispatchApplicationMessage(this, applicationMessage); + _sessionsManager.DispatchApplicationMessage(this, applicationMessage); return; } case MqttQualityOfServiceLevel.AtLeastOnce: { - _mqttClientSessionsManager.DispatchApplicationMessage(this, applicationMessage); + _sessionsManager.DispatchApplicationMessage(this, applicationMessage); - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }); return; @@ -218,9 +255,9 @@ namespace MQTTnet.Core.Server _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier); } - _mqttClientSessionsManager.DispatchApplicationMessage(this, applicationMessage); + _sessionsManager.DispatchApplicationMessage(this, applicationMessage); - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }); return; @@ -230,14 +267,14 @@ namespace MQTTnet.Core.Server } } - private Task HandleIncomingPubRelPacketAsync(IMqttCommunicationAdapter adapter, MqttPubRelPacket pubRelPacket) + private Task HandleIncomingPubRelPacketAsync(IMqttCommunicationAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { lock (_unacknowledgedPublishPackets) { _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier); } - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }); + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }); } } } diff --git a/MQTTnet.Core/Server/MqttClientSessionsManager.cs b/MQTTnet.Core/Server/MqttClientSessionsManager.cs index 09e07811..62cc1d82 100644 --- a/MQTTnet.Core/Server/MqttClientSessionsManager.cs +++ b/MQTTnet.Core/Server/MqttClientSessionsManager.cs @@ -11,7 +11,6 @@ using MQTTnet.Core.Protocol; using MQTTnet.Core.Serializer; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; -using MQTTnet.Core.Client; namespace MQTTnet.Core.Server { @@ -19,14 +18,19 @@ namespace MQTTnet.Core.Server { private readonly Dictionary _clientSessions = new Dictionary(); private readonly ILogger _logger; - private readonly IMqttClientSesssionFactory _mqttClientSesssionFactory; + private readonly IMqttClientSesssionFactory _clientSesssionFactory; + private readonly MqttServerOptions _options; - public MqttClientSessionsManager(IOptions options, ILogger logger, MqttClientRetainedMessagesManager retainedMessagesManager, IMqttClientSesssionFactory mqttClientSesssionFactory) + public MqttClientSessionsManager( + IOptions options, + ILogger logger, + MqttClientRetainedMessagesManager retainedMessagesManager, + IMqttClientSesssionFactory clientSesssionFactory) { _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - Options = options.Value ?? throw new ArgumentNullException(nameof(options)); + _options = options.Value ?? throw new ArgumentNullException(nameof(options)); RetainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(options)); - _mqttClientSesssionFactory = mqttClientSesssionFactory ?? throw new ArgumentNullException(nameof(mqttClientSesssionFactory)); + _clientSesssionFactory = clientSesssionFactory ?? throw new ArgumentNullException(nameof(clientSesssionFactory)); } public event EventHandler ClientConnected; @@ -34,14 +38,13 @@ namespace MQTTnet.Core.Server public event EventHandler ApplicationMessageReceived; public MqttClientRetainedMessagesManager RetainedMessagesManager { get; } - public MqttServerOptions Options { get; } - public async Task RunClientSessionAsync(IMqttCommunicationAdapter clientAdapter) + public async Task RunClientSessionAsync(IMqttCommunicationAdapter clientAdapter, CancellationToken cancellationToken) { var clientId = string.Empty; try { - if (!(await clientAdapter.ReceivePacketAsync(Options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false) is MqttConnectPacket connectPacket)) + if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false) is MqttConnectPacket connectPacket)) { throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1]."); } @@ -50,11 +53,11 @@ namespace MQTTnet.Core.Server // Switch to the required protocol version before sending any response. clientAdapter.PacketSerializer.ProtocolVersion = connectPacket.ProtocolVersion; - + var connectReturnCode = ValidateConnection(connectPacket); if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted) { - await clientAdapter.SendPacketsAsync(Options.DefaultCommunicationTimeout, CancellationToken.None, new MqttConnAckPacket + await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket { ConnectReturnCode = connectReturnCode }).ConfigureAwait(false); @@ -64,7 +67,7 @@ namespace MQTTnet.Core.Server var clientSession = GetOrCreateClientSession(connectPacket); - await clientAdapter.SendPacketsAsync(Options.DefaultCommunicationTimeout, CancellationToken.None, new MqttConnAckPacket + await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket { ConnectReturnCode = connectReturnCode, IsSessionPresent = clientSession.IsExistingSession @@ -89,7 +92,7 @@ namespace MQTTnet.Core.Server { try { - await clientAdapter.DisconnectAsync(Options.DefaultCommunicationTimeout).ConfigureAwait(false); + await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false); } catch (Exception) { @@ -147,9 +150,9 @@ namespace MQTTnet.Core.Server private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket) { - if (Options.ConnectionValidator != null) + if (_options.ConnectionValidator != null) { - return Options.ConnectionValidator(connectPacket); + return _options.ConnectionValidator(connectPacket); } return MqttConnectReturnCode.ConnectionAccepted; @@ -181,7 +184,7 @@ namespace MQTTnet.Core.Server { isExistingSession = false; - clientSession = _mqttClientSesssionFactory.CreateClientSession(connectPacket.ClientId, this); + clientSession = _clientSesssionFactory.CreateClientSession(connectPacket.ClientId, this); _clientSessions[connectPacket.ClientId] = clientSession; _logger.LogTrace("Created a new session for client '{0}'.", connectPacket.ClientId); diff --git a/MQTTnet.Core/Server/MqttClientSubscribeResult.cs b/MQTTnet.Core/Server/MqttClientSubscribeResult.cs new file mode 100644 index 00000000..17bfe9a5 --- /dev/null +++ b/MQTTnet.Core/Server/MqttClientSubscribeResult.cs @@ -0,0 +1,11 @@ +using MQTTnet.Core.Packets; + +namespace MQTTnet.Core.Server +{ + public class MqttClientSubscribeResult + { + public MqttSubAckPacket ResponsePacket { get; set; } + + public bool CloseConnection { get; set; } + } +} diff --git a/MQTTnet.Core/Server/MqttClientSubscriptionsManager.cs b/MQTTnet.Core/Server/MqttClientSubscriptionsManager.cs index 390f8f35..1b0d6c4c 100644 --- a/MQTTnet.Core/Server/MqttClientSubscriptionsManager.cs +++ b/MQTTnet.Core/Server/MqttClientSubscriptionsManager.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using Microsoft.Extensions.Options; using MQTTnet.Core.Packets; using MQTTnet.Core.Protocol; @@ -8,23 +9,45 @@ namespace MQTTnet.Core.Server public sealed class MqttClientSubscriptionsManager { private readonly Dictionary _subscribedTopics = new Dictionary(); + private readonly MqttServerOptions _options; - public MqttSubAckPacket Subscribe(MqttSubscribePacket subscribePacket) + public MqttClientSubscriptionsManager(IOptions options) + { + _options = options?.Value ?? throw new ArgumentNullException(nameof(options)); + } + + public MqttClientSubscribeResult Subscribe(MqttSubscribePacket subscribePacket, string clientId) { if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); var responsePacket = subscribePacket.CreateResponse(); + var closeConnection = false; lock (_subscribedTopics) { foreach (var topicFilter in subscribePacket.TopicFilters) { - _subscribedTopics[topicFilter.Topic] = topicFilter.QualityOfServiceLevel; - responsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS1); // TODO: Add support for QoS 2. + var interceptorContext = new MqttSubscriptionInterceptorContext(clientId, topicFilter); + _options.SubscriptionsInterceptor?.Invoke(interceptorContext); + responsePacket.SubscribeReturnCodes.Add(interceptorContext.AcceptSubscription ? MqttSubscribeReturnCode.SuccessMaximumQoS1 : MqttSubscribeReturnCode.Failure); + + if (interceptorContext.CloseConnection) + { + closeConnection = true; + } + + if (interceptorContext.AcceptSubscription) + { + _subscribedTopics[topicFilter.Topic] = topicFilter.QualityOfServiceLevel; + } } } - return responsePacket; + return new MqttClientSubscribeResult + { + ResponsePacket = responsePacket, + CloseConnection = closeConnection + }; } public MqttUnsubAckPacket Unsubscribe(MqttUnsubscribePacket unsubscribePacket) diff --git a/MQTTnet.Core/Server/MqttServer.cs b/MQTTnet.Core/Server/MqttServer.cs index 16e2e2ed..f1ed791e 100644 --- a/MQTTnet.Core/Server/MqttServer.cs +++ b/MQTTnet.Core/Server/MqttServer.cs @@ -6,7 +6,6 @@ using MQTTnet.Core.Adapter; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using System.Linq; -using MQTTnet.Core.Client; namespace MQTTnet.Core.Server { @@ -46,7 +45,7 @@ namespace MQTTnet.Core.Server public event EventHandler ClientDisconnected; public event EventHandler ApplicationMessageReceived; - public Task PublishAsync(IEnumerable applicationMessages) + public void Publish(IEnumerable applicationMessages) { if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); @@ -57,10 +56,20 @@ namespace MQTTnet.Core.Server foreach (var applicationMessage in applicationMessages) { - _options.ApplicationMessageInterceptor?.Invoke(applicationMessage); - _clientSessionsManager.DispatchApplicationMessage(null, applicationMessage); + var interceptorContext = new MqttApplicationMessageInterceptorContext + { + ApplicationMessage = applicationMessage + }; + + _options.ApplicationMessageInterceptor?.Invoke(interceptorContext); + + _clientSessionsManager.DispatchApplicationMessage(null, interceptorContext.ApplicationMessage); } - + } + + public Task PublishAsync(IEnumerable applicationMessages) + { + Publish(applicationMessages); return Task.FromResult(0); } @@ -100,7 +109,7 @@ namespace MQTTnet.Core.Server private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs) { - eventArgs.SessionTask = Task.Run(async () => await _clientSessionsManager.RunClientSessionAsync(eventArgs.Client), _cancellationTokenSource.Token); + eventArgs.SessionTask = Task.Run(async () => await _clientSessionsManager.RunClientSessionAsync(eventArgs.Client, _cancellationTokenSource.Token), _cancellationTokenSource.Token); } private void OnClientConnected(object sender, MqttClientConnectedEventArgs eventArgs) diff --git a/MQTTnet.Core/Server/MqttServerOptions.cs b/MQTTnet.Core/Server/MqttServerOptions.cs index 25f7cc8e..4b096764 100644 --- a/MQTTnet.Core/Server/MqttServerOptions.cs +++ b/MQTTnet.Core/Server/MqttServerOptions.cs @@ -16,7 +16,9 @@ namespace MQTTnet.Core.Server public Func ConnectionValidator { get; set; } - public Func ApplicationMessageInterceptor { get; set; } + public Action ApplicationMessageInterceptor { get; set; } + + public Action SubscriptionsInterceptor { get; set; } public IMqttServerStorage Storage { get; set; } } diff --git a/MQTTnet.Core/Server/MqttSubscribeResult.cs b/MQTTnet.Core/Server/MqttSubscribeResult.cs new file mode 100644 index 00000000..ef80af44 --- /dev/null +++ b/MQTTnet.Core/Server/MqttSubscribeResult.cs @@ -0,0 +1,11 @@ +using MQTTnet.Core.Packets; + +namespace MQTTnet.Core.Server +{ + public class MqttSubscribeResult + { + public MqttSubAckPacket ResponsePacket { get; set; } + + public bool CloseConnection { get; set; } + } +} diff --git a/MQTTnet.Core/Server/MqttSubscriptionInterceptorContext.cs b/MQTTnet.Core/Server/MqttSubscriptionInterceptorContext.cs new file mode 100644 index 00000000..93282d3b --- /dev/null +++ b/MQTTnet.Core/Server/MqttSubscriptionInterceptorContext.cs @@ -0,0 +1,19 @@ +namespace MQTTnet.Core.Server +{ + public class MqttSubscriptionInterceptorContext + { + public MqttSubscriptionInterceptorContext(string clientId, TopicFilter topicFilter) + { + ClientId = clientId; + TopicFilter = topicFilter; + } + + public string ClientId { get; } + + public TopicFilter TopicFilter { get; } + + public bool AcceptSubscription { get; set; } = true; + + public bool CloseConnection { get; set; } + } +} diff --git a/MQTTnet.sln b/MQTTnet.sln index 263fee0e..2207c1f8 100644 --- a/MQTTnet.sln +++ b/MQTTnet.sln @@ -1,12 +1,10 @@  Microsoft Visual Studio Solution File, Format Version 12.00 # Visual Studio 15 -VisualStudioVersion = 15.0.27004.2005 +VisualStudioVersion = 15.0.27004.2002 MinimumVisualStudioVersion = 10.0.40219.1 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Core.Tests", "Tests\MQTTnet.Core.Tests\MQTTnet.Core.Tests.csproj", "{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.UniversalWindows", "Frameworks\MQTTnet.UniversalWindows\MQTTnet.UniversalWindows.csproj", "{BD60C727-D8E8-40C3-B8E3-C95A864AE611}" -EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Tests", "Tests", "{9248C2E1-B9D6-40BF-81EC-86004D7765B4}" EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Frameworks", "Frameworks", "{32A630A7-2598-41D7-B625-204CD906F5FB}" @@ -19,7 +17,9 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.NetStandard", "Fram EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{67C28AC1-BC3A-420A-BE9C-FA2401431CF9}" ProjectSection(SolutionItems) = preProject + appveyor.yml = appveyor.yml Build\build.ps1 = Build\build.ps1 + Build\MQTTnet.AspNetCore.nuspec = Build\MQTTnet.AspNetCore.nuspec Build\MQTTnet.nuspec = Build\MQTTnet.nuspec EndProjectSection EndProject @@ -33,7 +33,7 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.TestApp.NetCore", " EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.TestApp.AspNetCore2", "Tests\MQTTnet.TestApp.AspNetCore2\MQTTnet.TestApp.AspNetCore2.csproj", "{C6FF8AEA-0855-41EC-A1F3-AC262225BAB9}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.AspnetCore", "Frameworks\MQTTnet.AspnetCore\MQTTnet.AspnetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Frameworks\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}" EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution @@ -63,22 +63,6 @@ Global {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x64.Build.0 = Release|Any CPU {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.ActiveCfg = Release|Any CPU {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.Build.0 = Release|Any CPU - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|Any CPU.Build.0 = Debug|Any CPU - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|ARM.ActiveCfg = Debug|ARM - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|ARM.Build.0 = Debug|ARM - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|x64.ActiveCfg = Debug|x64 - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|x64.Build.0 = Debug|x64 - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|x86.ActiveCfg = Debug|x86 - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|x86.Build.0 = Debug|x86 - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|Any CPU.ActiveCfg = Release|Any CPU - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|Any CPU.Build.0 = Release|Any CPU - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|ARM.ActiveCfg = Release|ARM - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|ARM.Build.0 = Release|ARM - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|x64.ActiveCfg = Release|x64 - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|x64.Build.0 = Release|x64 - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|x86.ActiveCfg = Release|x86 - {BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|x86.Build.0 = Release|x86 {2ECB99E4-72D0-4C23-99BA-93D511D3967D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {2ECB99E4-72D0-4C23-99BA-93D511D3967D}.Debug|Any CPU.Build.0 = Debug|Any CPU {2ECB99E4-72D0-4C23-99BA-93D511D3967D}.Debug|ARM.ActiveCfg = Debug|Any CPU @@ -187,7 +171,6 @@ Global EndGlobalSection GlobalSection(NestedProjects) = preSolution {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4} - {BD60C727-D8E8-40C3-B8E3-C95A864AE611} = {32A630A7-2598-41D7-B625-204CD906F5FB} {FF1F72D6-9524-4422-9497-3CC0002216ED} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4} {3587E506-55A2-4EB3-99C7-DC01E42D25D2} = {32A630A7-2598-41D7-B625-204CD906F5FB} {3D283AAD-AAA8-4339-8394-52F80B6304DB} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4} diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index d548ff08..3f5b4cf6 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -51,23 +51,33 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_WillMessage() { var serverAdapter = new TestMqttServerAdapter(); - var s = new MqttFactory().CreateMqttServer(); - await s.StartAsync(); - - var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build(); - var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - var c2 = await serverAdapter.ConnectTestClient(s, "c2", willMessage); + var services = new ServiceCollection() + .AddLogging() + .AddMqttServer() + .AddSingleton(serverAdapter) + .BuildServiceProvider(); + var s = new MqttFactory(services).CreateMqttServer(); var receivedMessagesCount = 0; - c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await c1.SubscribeAsync(new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce)); + try + { + await s.StartAsync(); - await c2.DisconnectAsync(); + var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build(); + var c1 = await serverAdapter.ConnectTestClient(s, "c1"); + var c2 = await serverAdapter.ConnectTestClient(s, "c2", willMessage); - await Task.Delay(1000); + c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; + await c1.SubscribeAsync(new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce)); - await s.StopAsync(); + await c2.DisconnectAsync(); + await Task.Delay(1000); + } + finally + { + await s.StopAsync(); + } Assert.AreEqual(1, receivedMessagesCount); } @@ -75,33 +85,45 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_Unsubscribe() { var serverAdapter = new TestMqttServerAdapter(); - var s = new MqttFactory().CreateMqttServer(); - await s.StartAsync(); + var services = new ServiceCollection() + .AddLogging() + .AddMqttServer() + .AddSingleton(serverAdapter) + .BuildServiceProvider(); - var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - var c2 = await serverAdapter.ConnectTestClient(s, "c2"); + var s = new MqttFactory(services).CreateMqttServer(); var receivedMessagesCount = 0; - c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); + try + { + await s.StartAsync(); - await c2.PublishAsync(message); - Assert.AreEqual(0, receivedMessagesCount); + var c1 = await serverAdapter.ConnectTestClient(s, "c1"); + var c2 = await serverAdapter.ConnectTestClient(s, "c2"); + c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce)); - await c2.PublishAsync(message); + var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); - await Task.Delay(500); - Assert.AreEqual(1, receivedMessagesCount); + await c2.PublishAsync(message); + Assert.AreEqual(0, receivedMessagesCount); - await c1.UnsubscribeAsync("a"); - await c2.PublishAsync(message); + await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce)); + await c2.PublishAsync(message); - await Task.Delay(500); - Assert.AreEqual(1, receivedMessagesCount); + await Task.Delay(500); + Assert.AreEqual(1, receivedMessagesCount); - await s.StopAsync(); + await c1.UnsubscribeAsync("a"); + await c2.PublishAsync(message); + + await Task.Delay(500); + Assert.AreEqual(1, receivedMessagesCount); + } + finally + { + await s.StopAsync(); + } await Task.Delay(500); Assert.AreEqual(1, receivedMessagesCount); @@ -111,22 +133,34 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_Publish() { var serverAdapter = new TestMqttServerAdapter(); - var s = new MqttFactory().CreateMqttServer(); - await s.StartAsync(); - - var c1 = await serverAdapter.ConnectTestClient(s, "c1"); + var services = new ServiceCollection() + .AddLogging() + .AddMqttServer() + .AddSingleton(serverAdapter) + .BuildServiceProvider(); + var s = new MqttFactory(services).CreateMqttServer(); var receivedMessagesCount = 0; - c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); - await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce)); + try + { + await s.StartAsync(); - s.PublishAsync(message).Wait(); - await Task.Delay(500); + var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - await s.StopAsync(); + c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; + var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); + await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce)); + + s.PublishAsync(message).Wait(); + await Task.Delay(500); + } + finally + { + await s.StopAsync(); + } + Assert.AreEqual(1, receivedMessagesCount); } @@ -134,21 +168,33 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_NoRetainedMessage() { var serverAdapter = new TestMqttServerAdapter(); - var s = new MqttFactory().CreateMqttServer(); - await s.StartAsync(); + var services = new ServiceCollection() + .AddLogging() + .AddMqttServer() + .AddSingleton(serverAdapter) + .BuildServiceProvider(); - var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).Build()); - await c1.DisconnectAsync(); - - var c2 = await serverAdapter.ConnectTestClient(s, "c2"); + var s = new MqttFactory(services).CreateMqttServer(); var receivedMessagesCount = 0; - c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); - await Task.Delay(500); + try + { + await s.StartAsync(); - await s.StopAsync(); + var c1 = await serverAdapter.ConnectTestClient(s, "c1"); + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).Build()); + await c1.DisconnectAsync(); + + var c2 = await serverAdapter.ConnectTestClient(s, "c2"); + c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; + await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); + + await Task.Delay(500); + } + finally + { + await s.StopAsync(); + } Assert.AreEqual(0, receivedMessagesCount); } @@ -157,22 +203,34 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_RetainedMessage() { var serverAdapter = new TestMqttServerAdapter(); - var s = new MqttFactory().CreateMqttServer(); - await s.StartAsync(); + var services = new ServiceCollection() + .AddLogging() + .AddMqttServer() + .AddSingleton(serverAdapter) + .BuildServiceProvider(); - var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); - await c1.DisconnectAsync(); + var s = new MqttFactory(services).CreateMqttServer(); - var c2 = await serverAdapter.ConnectTestClient(s, "c2"); var receivedMessagesCount = 0; - c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); + try + { + await s.StartAsync(); - await Task.Delay(500); + var c1 = await serverAdapter.ConnectTestClient(s, "c1"); + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); + await c1.DisconnectAsync(); - await s.StopAsync(); + var c2 = await serverAdapter.ConnectTestClient(s, "c2"); + c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; + await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); + await Task.Delay(500); + } + finally + { + await s.StopAsync(); + } + Assert.AreEqual(1, receivedMessagesCount); } @@ -182,26 +240,32 @@ namespace MQTTnet.Core.Tests var serverAdapter = new TestMqttServerAdapter(); var services = new ServiceCollection() .AddLogging() - .AddMqttServer() // TODO: Is there maybe an easier way for the library user to set the options? + .AddMqttServer() .AddSingleton(serverAdapter) .BuildServiceProvider(); var s = new MqttFactory(services).CreateMqttServer(); - await s.StartAsync(); - - var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); - await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build()); - await c1.DisconnectAsync(); - - var c2 = await serverAdapter.ConnectTestClient(s, "c2"); var receivedMessagesCount = 0; - c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); + try + { + await s.StartAsync(); - await Task.Delay(500); + var c1 = await serverAdapter.ConnectTestClient(s, "c1"); + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build()); + await c1.DisconnectAsync(); + + var c2 = await serverAdapter.ConnectTestClient(s, "c2"); + c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; + await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); + + await Task.Delay(500); + } + finally + { + await s.StopAsync(); + } - await s.StopAsync(); Assert.AreEqual(0, receivedMessagesCount); } @@ -214,30 +278,42 @@ namespace MQTTnet.Core.Tests var serverAdapter = new TestMqttServerAdapter(); var services = new ServiceCollection() .AddLogging() - .AddMqttServer(options => options.Storage = storage) // TODO: Is there maybe an easier way for the library user to set the options? + .AddMqttServer() .AddSingleton(serverAdapter) .BuildServiceProvider(); - var s = new MqttFactory(services).CreateMqttServer(); // TODO: Like here? - await s.StartAsync(); + var s = new MqttFactory(services).CreateMqttServer(options => options.Storage = storage); - var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); - await c1.DisconnectAsync(); + try + { + await s.StartAsync(); - await s.StopAsync(); + var c1 = await serverAdapter.ConnectTestClient(s, "c1"); + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); + await c1.DisconnectAsync(); + } + finally + { + await s.StopAsync(); + } s = services.GetRequiredService(); - await s.StartAsync(); - var c2 = await serverAdapter.ConnectTestClient(s, "c2"); var receivedMessagesCount = 0; - c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); + try + { + await s.StartAsync(); - await Task.Delay(500); + var c2 = await serverAdapter.ConnectTestClient(s, "c2"); + c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; + await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); - await s.StopAsync(); + await Task.Delay(500); + } + finally + { + await s.StopAsync(); + } Assert.AreEqual(1, receivedMessagesCount); } @@ -245,39 +321,45 @@ namespace MQTTnet.Core.Tests [TestMethod] public async Task MqttServer_InterceptMessage() { - MqttApplicationMessage Interceptor(MqttApplicationMessage message) + void Interceptor(MqttApplicationMessageInterceptorContext context) { - message.Payload = Encoding.ASCII.GetBytes("extended"); - return message; + context.ApplicationMessage.Payload = Encoding.ASCII.GetBytes("extended"); } var serverAdapter = new TestMqttServerAdapter(); var services = new ServiceCollection() .AddLogging() - .AddMqttServer(options => options.ApplicationMessageInterceptor = Interceptor) + .AddMqttServer() .AddSingleton(serverAdapter) .BuildServiceProvider(); - var s = services.GetRequiredService(); - await s.StartAsync(); - - var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - var c2 = await serverAdapter.ConnectTestClient(s, "c2"); - await c2.SubscribeAsync(new TopicFilterBuilder().WithTopic("test").Build()); - - var isIntercepted = false; - c2.ApplicationMessageReceived += (sender, args) => + var s = new MqttFactory(services).CreateMqttServer(options => options.ApplicationMessageInterceptor = Interceptor); + try { - isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(args.ApplicationMessage.Payload), StringComparison.Ordinal) == 0; - }; + await s.StartAsync(); - var m = new MqttApplicationMessageBuilder().WithTopic("test").Build(); - await c1.PublishAsync(m); - await c1.DisconnectAsync(); + var c1 = await serverAdapter.ConnectTestClient(s, "c1"); + var c2 = await serverAdapter.ConnectTestClient(s, "c2"); + await c2.SubscribeAsync(new TopicFilterBuilder().WithTopic("test").Build()); - await Task.Delay(500); + var isIntercepted = false; + c2.ApplicationMessageReceived += (sender, args) => + { + isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(args.ApplicationMessage.Payload), StringComparison.Ordinal) == 0; + }; - Assert.IsTrue(isIntercepted); + var m = new MqttApplicationMessageBuilder().WithTopic("test").Build(); + await c1.PublishAsync(m); + await c1.DisconnectAsync(); + + await Task.Delay(500); + + Assert.IsTrue(isIntercepted); + } + finally + { + await s.StopAsync(); + } } private class TestStorage : IMqttServerStorage @@ -306,28 +388,34 @@ namespace MQTTnet.Core.Tests var serverAdapter = new TestMqttServerAdapter(); var services = new ServiceCollection() .AddMqttServer() + .AddLogging() .AddSingleton(serverAdapter) .BuildServiceProvider(); var s = services.GetRequiredService(); - await s.StartAsync(); - - var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - var c2 = await serverAdapter.ConnectTestClient(s, "c2"); - var receivedMessagesCount = 0; - c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; + try + { + await s.StartAsync(); - await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic(topicFilter).WithQualityOfServiceLevel(filterQualityOfServiceLevel).Build()); - await c2.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(new byte[0]).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); + var c1 = await serverAdapter.ConnectTestClient(s, "c1"); + var c2 = await serverAdapter.ConnectTestClient(s, "c2"); - await Task.Delay(500); - await c1.UnsubscribeAsync(topicFilter); + c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await Task.Delay(500); + await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic(topicFilter).WithQualityOfServiceLevel(filterQualityOfServiceLevel).Build()); + await c2.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(new byte[0]).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); - await s.StopAsync(); + await Task.Delay(500); + await c1.UnsubscribeAsync(topicFilter); + await Task.Delay(500); + } + finally + { + await s.StopAsync(); + } + Assert.AreEqual(expectedReceivedMessagesCount, receivedMessagesCount); } } diff --git a/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs b/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs index e4777317..d46354c7 100644 --- a/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs @@ -1,4 +1,5 @@ -using Microsoft.VisualStudio.TestTools.UnitTesting; +using Microsoft.Extensions.Options; +using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Core.Packets; using MQTTnet.Core.Protocol; using MQTTnet.Core.Server; @@ -11,12 +12,12 @@ namespace MQTTnet.Core.Tests [TestMethod] public void MqttSubscriptionsManager_SubscribeSingleSuccess() { - var sm = new MqttClientSubscriptionsManager(); + var sm = new MqttClientSubscriptionsManager(new OptionsWrapper(new MqttServerOptions())); var sp = new MqttSubscribePacket(); sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtMostOnce)); - sm.Subscribe(sp); + sm.Subscribe(sp, ""); var pp = new MqttPublishPacket { @@ -30,12 +31,12 @@ namespace MQTTnet.Core.Tests [TestMethod] public void MqttSubscriptionsManager_SubscribeSingleNoSuccess() { - var sm = new MqttClientSubscriptionsManager(); + var sm = new MqttClientSubscriptionsManager(new OptionsWrapper(new MqttServerOptions())); var sp = new MqttSubscribePacket(); sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtMostOnce)); - sm.Subscribe(sp); + sm.Subscribe(sp, ""); var pp = new MqttPublishPacket { @@ -49,12 +50,12 @@ namespace MQTTnet.Core.Tests [TestMethod] public void MqttSubscriptionsManager_SubscribeAndUnsubscribeSingle() { - var sm = new MqttClientSubscriptionsManager(); + var sm = new MqttClientSubscriptionsManager(new OptionsWrapper(new MqttServerOptions())); var sp = new MqttSubscribePacket(); sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtMostOnce)); - sm.Subscribe(sp); + sm.Subscribe(sp, ""); var pp = new MqttPublishPacket { diff --git a/Tests/MQTTnet.Core.Tests/TestClientSessionFactory.cs b/Tests/MQTTnet.Core.Tests/TestClientSessionFactory.cs index 519ef500..638df85d 100644 --- a/Tests/MQTTnet.Core.Tests/TestClientSessionFactory.cs +++ b/Tests/MQTTnet.Core.Tests/TestClientSessionFactory.cs @@ -1,12 +1,14 @@ -using MQTTnet.Core.Server; +using System; +using MQTTnet.Core.Server; namespace MQTTnet.Core.Tests { public class TestClientSessionFactory : IMqttClientSesssionFactory { - public MqttClientSession CreateClientSession(string sessionId, MqttClientSessionsManager mqttClientSessionsManager) + public MqttClientSession CreateClientSession(string clientId, MqttClientSessionsManager mqttClientSessionsManager) { - return new MqttClientSession(sessionId, mqttClientSessionsManager, new TestLogger(), new TestLogger()); + throw new NotImplementedException(); + //return new MqttClientSession(clientId, mqttClientSessionsManager, new TestLogger(), new TestLogger()); } } } diff --git a/Tests/MQTTnet.Core.Tests/TestLogger.cs b/Tests/MQTTnet.Core.Tests/TestLogger.cs index ea597a6e..f57c877c 100644 --- a/Tests/MQTTnet.Core.Tests/TestLogger.cs +++ b/Tests/MQTTnet.Core.Tests/TestLogger.cs @@ -3,11 +3,11 @@ using System; namespace MQTTnet.Core.Tests { - public class TestLogger : ILogger + public class TestLogger : IDisposable, ILogger { public IDisposable BeginScope(TState state) { - throw new NotImplementedException(); + return this; } public bool IsEnabled(LogLevel logLevel) @@ -18,5 +18,9 @@ namespace MQTTnet.Core.Tests public void Log(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func formatter) { } + + public void Dispose() + { + } } } diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj index 2f042084..7e4e0e3d 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj +++ b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj @@ -14,7 +14,7 @@ - + diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs index c5da4e5a..18d121cf 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs @@ -6,7 +6,7 @@ using Microsoft.AspNetCore.Hosting; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.FileProviders; using Microsoft.Extensions.Logging; -using MQTTnet.AspnetCore; +using MQTTnet.AspNetCore; using MQTTnet.Core; using MQTTnet.Core.Client; diff --git a/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs b/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs index a02e72c4..7f6d430b 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs @@ -50,14 +50,14 @@ namespace MQTTnet.TestApp.NetCore Console.WriteLine(">> RECEIVED: " + e.ApplicationMessage.Topic); }; - await managedClient.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic("Step").WithPayload("1").Build()); - await managedClient.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic("Step").WithPayload("2").WithAtLeastOnceQoS().Build()); + await managedClient.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("Step").WithPayload("1").Build()); + await managedClient.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("Step").WithPayload("2").WithAtLeastOnceQoS().Build()); await managedClient.StartAsync(options); await managedClient.SubscribeAsync(new TopicFilter("xyz", MqttQualityOfServiceLevel.AtMostOnce)); - await managedClient.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic("Step").WithPayload("3").Build()); + await managedClient.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("Step").WithPayload("3").Build()); Console.WriteLine("Managed client started."); Console.ReadLine(); diff --git a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs index a0c64623..0a7d5575 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs @@ -35,16 +35,14 @@ namespace MQTTnet.TestApp.NetCore options.Storage = new RetainedMessageHandler(); - options.ApplicationMessageInterceptor = message => + options.ApplicationMessageInterceptor = context => { - if (MqttTopicFilterComparer.IsMatch(message.Topic, "/myTopic/WithTimestamp/#")) + if (MqttTopicFilterComparer.IsMatch(context.ApplicationMessage.Topic, "/myTopic/WithTimestamp/#")) { // Replace the payload with the timestamp. But also extending a JSON // based payload with the timestamp is a suitable use case. - message.Payload = Encoding.UTF8.GetBytes(DateTime.Now.ToString("O")); - } - - return message; + context.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(DateTime.Now.ToString("O")); + } }; }); diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj index e7c65f20..42b16304 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj @@ -18,7 +18,7 @@ {A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC} true MQTTnet.TestApp.UniversalWindows_TemporaryKey.pfx - 13E377A693C923EE9E88EE84B32A4B9881657362 + FA56408F253A1F7FCAC649611411B6B9F42BC4F9 win10;win10-arm;win10-arm-aot;win10-x86;win10-x86-aot;win10-x64;win10-x64-aot @@ -126,9 +126,9 @@ - - {bd60c727-d8e8-40c3-b8e3-c95a864ae611} - MQTTnet.UniversalWindows + + {3587e506-55a2-4eb3-99c7-dc01e42d25d2} + MQTTnet.NetStandard {2ecb99e4-72d0-4c23-99ba-93d511d3967d} diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows_TemporaryKey.pfx b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows_TemporaryKey.pfx new file mode 100644 index 00000000..adf1e2aa Binary files /dev/null and b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows_TemporaryKey.pfx differ diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml index e2b0a187..f0031b8b 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml @@ -65,7 +65,6 @@ - Topic: @@ -84,6 +83,17 @@ + + + Port: + + + + + + + + diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs index 3663ba4c..76c0d073 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs @@ -9,6 +9,7 @@ using MQTTnet.Core; using MQTTnet.Core.Client; using MQTTnet.Core.Diagnostics; using MQTTnet.Core.Protocol; +using MQTTnet.Core.Server; using MQTTnet.Implementations; namespace MQTTnet.TestApp.UniversalWindows @@ -16,6 +17,7 @@ namespace MQTTnet.TestApp.UniversalWindows public sealed partial class MainPage { private IMqttClient _mqttClient; + private IMqttServer _mqttServer; public MainPage() { @@ -319,5 +321,31 @@ namespace MQTTnet.TestApp.UniversalWindows }; } + + private async void StartServer(object sender, RoutedEventArgs e) + { + if (_mqttServer != null) + { + return; + } + + _mqttServer = new MqttFactory().CreateMqttServer(o => + { + o.DefaultEndpointOptions.Port = int.Parse(ServerPort.Text); + }); + + await _mqttServer.StartAsync(); + } + + private async void StopServer(object sender, RoutedEventArgs e) + { + if (_mqttServer == null) + { + return; + } + + await _mqttServer.StopAsync(); + _mqttServer = null; + } } } diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/Package.appxmanifest b/Tests/MQTTnet.TestApp.UniversalWindows/Package.appxmanifest index d6a7c4c1..a3094c70 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/Package.appxmanifest +++ b/Tests/MQTTnet.TestApp.UniversalWindows/Package.appxmanifest @@ -1,6 +1,6 @@  - + MQTTnet.TestApp.UniversalWindows diff --git a/appveyor.yml b/appveyor.yml new file mode 100644 index 00000000..97816dd1 --- /dev/null +++ b/appveyor.yml @@ -0,0 +1,15 @@ +version: 1.0.{build} +image: Visual Studio 2017 +before_build: +- cmd: >- + msbuild /t:restore + + cd Tests/MQTTnet.TestApp.AspNetCore2/ + + npm install + +build: + project: MQTTnet.sln + verbosity: minimal +test_script: +- cmd: vstest.console /logger:Appveyor Tests/MQTTnet.Core.Tests/bin/Release/net461/MQTTnet.Core.Tests.dll