YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-17 09:43:05 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3 from chkr1011/develop

Browse Source 
Original Develop
This commit is contained in:
Gerardo
2017-10-30 11:08:41 +01:00
committed by GitHub
parent e6d811ea7c c8846329f3
commit d076aac421
58 changed files with 665 additions and 876 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
-1
View File
@@ -197,7 +197,6 @@ ClientBin/
*.dbmdl
*.dbproj.schemaview
*.jfm
*.pfx
*.publishsettings
orleans.codegen.cs
Build/MQTTnet.AspNetCore.nuspec
+30
View File
@@ -0,0 +1,30 @@
<?xml version="1.0"?>
<package >
<metadata>
<id>MQTTnet.AspNetCore</id>
<version>2.5.0</version>
<authors>Christian Kratky</authors>
<owners>Christian Kratky</owners>
<licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
<projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<requireLicenseAcceptance>false</requireLicenseAcceptance>
<description>This is a support library to integrate MQTTnet into AspNetCore.</description>
<releaseNotes>initial version
</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2017</copyright>
<tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation</tags>
<dependencies>
<group targetFramework="netstandard2.0">
<dependency id="MQTTnet" version="2.5" />
</group>
</dependencies>
</metadata>
<files>
<!-- .NET Standard 2.0 -->
<file src="..\Frameworks\MQTTnet.AspNetCore\bin\Release\netstandard2.0\MQTTnet.AspNetCore.*" target="lib\netstandard2.0\"/>
</files>
</package>
Build/MQTTnet.nuspec
+4 -4
View File
@@ -47,12 +47,12 @@
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard1.3\MQTTnet.*" target="lib\netstandard1.3\"/>
<!-- Universal Windows -->
<file src="..\Frameworks\MQTTnet.UniversalWindows\bin\Release\MQTTnet.Core.*" target="lib\uap10.0\"/>
<file src="..\Frameworks\MQTTnet.UniversalWindows\bin\Release\MQTTnet.*" target="lib\uap10.0\"/>
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\uap10.0\MQTTnet.Core.*" target="lib\uap10.0\"/>
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\uap10.0\MQTTnet.*" target="lib\uap10.0\"/>
<!-- .NET Framework -->
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net45\MQTTnet.Core.*" target="lib\net45\"/>
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net45\MQTTnet.*" target="lib\net45\"/>
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net451\MQTTnet.Core.*" target="lib\net451\"/>
<file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net451\MQTTnet.*" target="lib\net451\"/>
</files>
</package>
Build/build.ps1
+16 -6
View File
@@ -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
Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs
+1 -1
View File
@@ -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
{
Frameworks/MQTTnet.AspnetCore/MqttHostedServer.cs
+1 -1
View File
@@ -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
{
Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs
+1 -1
View File
@@ -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
{
Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs
+1 -1
View File
@@ -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
{
Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs → Frameworks/MQTTnet.NetStandard/Implementations/NetStandard/MqttServerAdapter.cs
+5 -2
View File
@@ -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
}
}
}
}
}
#endif
Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs → Frameworks/MQTTnet.NetStandard/Implementations/NetStandard/MqttTcpChannel.cs
+4 -1
View File
@@ -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
}
}
}
}
#endif
Frameworks/MQTTnet.UniversalWindows/Implementations/MqttServerAdapter.cs → Frameworks/MQTTnet.NetStandard/Implementations/Uap/MqttServerAdapter.cs
+7 -1
View File
@@ -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
}
}
}
}
}
#endif
Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs → Frameworks/MQTTnet.NetStandard/Implementations/Uap/MqttTcpChannel.cs
+4 -1
View File
@@ -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;
}
}
}
}
#endif
Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj
+18 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>netstandard1.3;net451</TargetFrameworks>
<TargetFrameworks>netstandard1.3;net451;uap10.0</TargetFrameworks>
<AssemblyName>MQTTnet</AssemblyName>
<RootNamespace>MQTTnet</RootNamespace>
<AssemblyVersion>2.5.0.0</AssemblyVersion>
@@ -14,6 +14,19 @@
<PackageId />
</PropertyGroup>
<PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0'">
<CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
<NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
<TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
<TargetPlatformVersion>10.0.15063.0</TargetPlatformVersion>
<TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
<TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
<TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
<DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
<LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x86'" />
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'" />
@@ -34,4 +47,8 @@
<PackageReference Include="System.Threading.Thread" Version="4.3.0" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="5.4.0" />
</ItemGroup>
</Project>
Frameworks/MQTTnet.NetStandard/MqttFactory.cs
+19 -2
View File
@@ -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<ILogger<MqttClientSession>>(), _serviceProvider.GetRequiredService<ILogger<MqttClientPendingMessagesQueue>>());
return new MqttClientSession(
clientId,
_serviceProvider.GetRequiredService<IOptions<MqttServerOptions>>(),
clientSessionsManager,
_serviceProvider.GetRequiredService<MqttClientSubscriptionsManager>(),
_serviceProvider.GetRequiredService<ILogger<MqttClientSession>>(),
_serviceProvider.GetRequiredService<ILogger<MqttClientPendingMessagesQueue>>());
}
public IMqttClient CreateMqttClient()
@@ -109,5 +116,15 @@ namespace MQTTnet
{
return _serviceProvider.GetRequiredService<IMqttServer>();
}
public IMqttServer CreateMqttServer(Action<MqttServerOptions> configure)
{
if (configure == null) throw new ArgumentNullException(nameof(configure));
var options = _serviceProvider.GetRequiredService<IOptions<MqttServerOptions>>();
configure(options.Value);
return _serviceProvider.GetRequiredService<IMqttServer>();
}
}
}
Frameworks/MQTTnet.NetStandard/ServiceCollectionExtensions.cs
+1
View File
@@ -34,6 +34,7 @@ namespace MQTTnet
services.AddTransient<IMqttPacketSerializer, MqttPacketSerializer>();
services.AddTransient<MqttClientSessionsManager>();
services.AddTransient<MqttClientSubscriptionsManager>();
services.AddTransient<MqttClientRetainedMessagesManager>();
return services;
}
Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs
-92
View File
@@ -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;
}
}
}
Frameworks/MQTTnet.UniversalWindows/Implementations/WebSocketStream.cs
-84
View File
@@ -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<int> 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<byte>(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<byte>(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();
}
}
}
Frameworks/MQTTnet.UniversalWindows/MQTTnet.UniversalWindows.csproj
-200
View File
@@ -1,200 +0,0 @@
<?xml version="1.0" encoding="utf-8"?>
<Project ToolsVersion="14.0" DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
<Import Project="$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props" Condition="Exists('$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props')" />
<PropertyGroup>
<Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
<Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
<ProjectGuid>{BD60C727-D8E8-40C3-B8E3-C95A864AE611}</ProjectGuid>
<OutputType>Library</OutputType>
<AppDesignerFolder>Properties</AppDesignerFolder>
<RootNamespace>MQTTnet</RootNamespace>
<AssemblyName>MQTTnet</AssemblyName>
<DefaultLanguage>en-US</DefaultLanguage>
<TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
<TargetPlatformVersion>10.0.15063.0</TargetPlatformVersion>
<TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
<MinimumVisualStudioVersion>14</MinimumVisualStudioVersion>
<FileAlignment>512</FileAlignment>
<ProjectTypeGuids>{A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids>
<RuntimeIdentifiers>win10-arm;win10-arm-aot;win10-x86;win10-x86-aot;win10-x64;win10-x64-aot</RuntimeIdentifiers>
</PropertyGroup>
<PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
<PlatformTarget>AnyCPU</PlatformTarget>
<DebugSymbols>true</DebugSymbols>
<DebugType>full</DebugType>
<Optimize>false</Optimize>
<OutputPath>bin\Debug\</OutputPath>
<DefineConstants>DEBUG;TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<ErrorReport>prompt</ErrorReport>
<WarningLevel>4</WarningLevel>
<NoWarn>CS1591</NoWarn>
</PropertyGroup>
<PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
<PlatformTarget>AnyCPU</PlatformTarget>
<DebugType>pdbonly</DebugType>
<Optimize>true</Optimize>
<OutputPath>bin\Release\</OutputPath>
<DefineConstants>TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<ErrorReport>prompt</ErrorReport>
<WarningLevel>4</WarningLevel>
<NoWarn>CS1591</NoWarn>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Debug|x86'">
<PlatformTarget>x86</PlatformTarget>
<DebugSymbols>true</DebugSymbols>
<OutputPath>bin\x86\Debug\</OutputPath>
<DefineConstants>DEBUG;TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<NoWarn>;2008,CS1591</NoWarn>
<DebugType>full</DebugType>
<PlatformTarget>x86</PlatformTarget>
<UseVSHostingProcess>false</UseVSHostingProcess>
<ErrorReport>prompt</ErrorReport>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Release|x86'">
<PlatformTarget>x86</PlatformTarget>
<OutputPath>bin\x86\Release\</OutputPath>
<DefineConstants>TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<Optimize>true</Optimize>
<NoWarn>;2008,CS1591</NoWarn>
<DebugType>pdbonly</DebugType>
<PlatformTarget>x86</PlatformTarget>
<UseVSHostingProcess>false</UseVSHostingProcess>
<ErrorReport>prompt</ErrorReport>
<DocumentationFile>bin\x86\Release\MQTTnet.XML</DocumentationFile>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Debug|ARM'">
<PlatformTarget>ARM</PlatformTarget>
<DebugSymbols>true</DebugSymbols>
<OutputPath>bin\ARM\Debug\</OutputPath>
<DefineConstants>DEBUG;TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<NoWarn>;2008,CS1591</NoWarn>
<DebugType>full</DebugType>
<PlatformTarget>ARM</PlatformTarget>
<UseVSHostingProcess>false</UseVSHostingProcess>
<ErrorReport>prompt</ErrorReport>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Release|ARM'">
<PlatformTarget>ARM</PlatformTarget>
<OutputPath>bin\ARM\Release\</OutputPath>
<DefineConstants>TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<Optimize>true</Optimize>
<NoWarn>;2008,CS1591</NoWarn>
<DebugType>pdbonly</DebugType>
<PlatformTarget>ARM</PlatformTarget>
<UseVSHostingProcess>false</UseVSHostingProcess>
<ErrorReport>prompt</ErrorReport>
<DocumentationFile>bin\ARM\Release\MQTTnet.XML</DocumentationFile>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Debug|x64'">
<PlatformTarget>x64</PlatformTarget>
<DebugSymbols>true</DebugSymbols>
<OutputPath>bin\x64\Debug\</OutputPath>
<DefineConstants>DEBUG;TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<NoWarn>;2008,CS1591</NoWarn>
<DebugType>full</DebugType>
<PlatformTarget>x64</PlatformTarget>
<UseVSHostingProcess>false</UseVSHostingProcess>
<ErrorReport>prompt</ErrorReport>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Release|x64'">
<PlatformTarget>x64</PlatformTarget>
<OutputPath>bin\x64\Release\</OutputPath>
<DefineConstants>TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<Optimize>true</Optimize>
<NoWarn>;2008,CS1591</NoWarn>
<DebugType>pdbonly</DebugType>
<PlatformTarget>x64</PlatformTarget>
<UseVSHostingProcess>false</UseVSHostingProcess>
<ErrorReport>prompt</ErrorReport>
<GenerateLibraryLayout>false</GenerateLibraryLayout>
<DocumentationFile>bin\x64\Release\MQTTnet.XML</DocumentationFile>
</PropertyGroup>
<ItemGroup>
<Compile Include="Implementations\MqttWebSocketChannel.cs" />
<Compile Include="MqttFactory.cs" />
<Compile Include="Implementations\MqttServerAdapter.cs" />
<Compile Include="Implementations\MqttTcpChannel.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
<Compile Include="Implementations\WebSocketStream.cs" />
<Compile Include="ServiceCollectionExtensions.cs" />
<EmbeddedResource Include="Properties\MQTTnet.Universal.rd.xml" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj">
<Project>{2ecb99e4-72d0-4c23-99ba-93d511d3967d}</Project>
<Name>MQTTnet.Core</Name>
</ProjectReference>
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.DependencyInjection">
<Version>1.1.1</Version>
</PackageReference>
<PackageReference Include="Microsoft.Extensions.Logging">
<Version>1.1.2</Version>
</PackageReference>
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform">
<Version>5.4.0</Version>
</PackageReference>
</ItemGroup>
<PropertyGroup Condition=" '$(VisualStudioVersion)' == '' or '$(VisualStudioVersion)' &lt; '14.0' ">
<VisualStudioVersion>14.0</VisualStudioVersion>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Any CPU|AnyCPU'">
<OutputPath>bin\Any CPU\</OutputPath>
<DefineConstants>TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<Optimize>true</Optimize>
<NoStdLib>true</NoStdLib>
<DebugType>pdbonly</DebugType>
<PlatformTarget>AnyCPU</PlatformTarget>
<UseVSHostingProcess>false</UseVSHostingProcess>
<ErrorReport>prompt</ErrorReport>
<CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Any CPU|x86'">
<OutputPath>bin\x86\Any CPU\</OutputPath>
<DefineConstants>TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<DocumentationFile>bin\x86\Release\MQTTnet.XML</DocumentationFile>
<Optimize>true</Optimize>
<NoWarn>;2008</NoWarn>
<NoStdLib>true</NoStdLib>
<DebugType>pdbonly</DebugType>
<PlatformTarget>x86</PlatformTarget>
<UseVSHostingProcess>false</UseVSHostingProcess>
<ErrorReport>prompt</ErrorReport>
<CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Any CPU|ARM'">
<OutputPath>bin\ARM\Any CPU\</OutputPath>
<DefineConstants>TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<DocumentationFile>bin\ARM\Release\MQTTnet.XML</DocumentationFile>
<Optimize>true</Optimize>
<NoWarn>;2008</NoWarn>
<NoStdLib>true</NoStdLib>
<DebugType>pdbonly</DebugType>
<PlatformTarget>ARM</PlatformTarget>
<UseVSHostingProcess>false</UseVSHostingProcess>
<ErrorReport>prompt</ErrorReport>
<CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Any CPU|x64'">
<OutputPath>bin\x64\Any CPU\</OutputPath>
<DefineConstants>TRACE;NETFX_CORE;WINDOWS_UWP</DefineConstants>
<DocumentationFile>bin\x64\Release\MQTTnet.XML</DocumentationFile>
<Optimize>true</Optimize>
<NoWarn>;2008</NoWarn>
<NoStdLib>true</NoStdLib>
<DebugType>pdbonly</DebugType>
<PlatformTarget>x64</PlatformTarget>
<UseVSHostingProcess>false</UseVSHostingProcess>
<ErrorReport>prompt</ErrorReport>
<CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
</PropertyGroup>
<Import Project="$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets" />
<!-- To modify your build process, add your task inside one of the targets below and uncomment it.
Other similar extension points exist, see Microsoft.Common.targets.
<Target Name="BeforeBuild">
</Target>
<Target Name="AfterBuild">
</Target>
-->
</Project>
Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs
-108
View File
@@ -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<ILoggerFactory>()
.AddMqttTrace();
return serviceProvider;
}
public MqttFactory()
: this(BuildServiceProvider())
{
}
public MqttFactory(IServiceProvider serviceProvider)
{
_serviceProvider = serviceProvider;
}
public IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options)
{
var logger = _serviceProvider.GetRequiredService<ILogger<MqttChannelCommunicationAdapter>>();
return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options.ChannelOptions), CreateSerializer(options.ProtocolVersion), logger);
}
public IMqttCommunicationAdapter CreateServerMqttCommunicationAdapter(IMqttCommunicationChannel channel)
{
var serializer = _serviceProvider.GetRequiredService<IMqttPacketSerializer>();
var logger = _serviceProvider.GetRequiredService<ILogger<MqttChannelCommunicationAdapter>>();
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<ILogger<MqttClientSession>>(), _serviceProvider.GetRequiredService<ILogger<MqttClientPendingMessagesQueue>>());
}
public IMqttClient CreateMqttClient()
{
return _serviceProvider.GetRequiredService<IMqttClient>();
}
public IManagedMqttClient CreateManagedMqttClient()
{
return _serviceProvider.GetRequiredService<ManagedMqttClient>();
}
public IMqttServer CreateMqttServer()
{
return _serviceProvider.GetRequiredService<IMqttServer>();
}
}
}
Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs
-14
View File
@@ -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")]
Frameworks/MQTTnet.UniversalWindows/Properties/MQTTnet.Universal.rd.xml
-5
View File
@@ -1,5 +0,0 @@
<?xml version="1.0" encoding="utf-8"?>
<Directives xmlns="http://schemas.microsoft.com/netfx/2013/01/metadata">
<Library Name="MQTTnet.Universal">
</Library>
</Directives>
Frameworks/MQTTnet.UniversalWindows/ServiceCollectionExtensions.cs
-63
View File
@@ -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<MqttFactory>();
services.AddSingleton<IMqttCommunicationAdapterFactory, MqttFactory>();
services.AddSingleton<IMqttClientSesssionFactory, MqttFactory>();
services.AddSingleton<IMqttServer,MqttServer>();
services.AddSingleton<MqttServer>();
services.AddTransient<IMqttServerAdapter, MqttServerAdapter>();
services.AddTransient<IMqttPacketSerializer, MqttPacketSerializer>();
services.AddTransient<MqttClientSessionsManager>();
services.AddTransient<MqttClientRetainedMessagesManager>();
return services;
}
public static IServiceCollection AddMqttServer(this IServiceCollection services, Action<MqttServerOptions> configureOptions)
{
return services
.AddMqttServer()
.Configure(configureOptions);
}
public static IServiceCollection AddMqttClient(this IServiceCollection services)
{
services.AddSingleton<MqttFactory>();
services.AddSingleton<IMqttCommunicationAdapterFactory, MqttFactory>();
services.AddTransient<IMqttClient, MqttClient>();
services.AddTransient<MqttClient>();
services.AddTransient<IManagedMqttClient, ManagedMqttClient>();
services.AddTransient<ManagedMqttClient>();
services.AddTransient<IMqttPacketSerializer, MqttPacketSerializer>();
services.AddTransient<MqttPacketDispatcher>();
return services;
}
public static ILoggerFactory AddMqttTrace(this ILoggerFactory factory)
{
factory.AddProvider(new MqttNetTrace());
return factory;
}
}
}
MQTTnet.Core/ApplicationMessagePublisherExtensions.cs
+16
View File
@@ -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);
}
}
}
MQTTnet.Core/Client/IMqttClientOptions.cs
+5
View File
@@ -6,6 +6,11 @@ namespace MQTTnet.Core.Client
public interface IMqttClientOptions
{
string ClientId { get; }
/// <summary>
/// The LogId is used to create a scope to correlate logging. If no value is provided the ClientId is used instead
/// </summary>
string LogId { get; }
IMqttClientCredentials Credentials { get; }
bool CleanSession { get; }
MqttApplicationMessage WillMessage { get; }
MQTTnet.Core/Client/MqttClient.cs
+31 -30
View File
@@ -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<MqttPubAckPacket>(publishPacket);
}
break;
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket);
}
break;
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
foreach (var publishPacket in qosPackets)
{
foreach (var publishPacket in qosPackets)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>()).ConfigureAwait(false);
}
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>()).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
}
}
MQTTnet.Core/Client/MqttClientExtensions.cs
-8
View File
@@ -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<IList<MqttSubscribeResult>> SubscribeAsync(this IMqttClient client, params TopicFilter[] topicFilters)
{
if (client == null) throw new ArgumentNullException(nameof(client));
MQTTnet.Core/Client/MqttClientOptions.cs
+3
View File
@@ -9,6 +9,9 @@ namespace MQTTnet.Core.Client
public string ClientId { get; set; } = Guid.NewGuid().ToString("N");
/// <inheritdoc />
public string LogId { get; set; }
public bool CleanSession { get; set; } = true;
public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials();
MQTTnet.Core/Client/IApplicationMessagePublisher.cs → MQTTnet.Core/IApplicationMessagePublisher.cs
+1 -1
View File
@@ -1,7 +1,7 @@
using System.Collections.Generic;
using System.Threading.Tasks;
namespace MQTTnet.Core.Client
namespace MQTTnet.Core
{
public interface IApplicationMessagePublisher
{
MQTTnet.Core/Client/IApplicationMessageReceiver.cs → MQTTnet.Core/IApplicationMessageReceiver.cs
+2 -1
View File
@@ -1,6 +1,7 @@
using System;
using MQTTnet.Core.Client;
namespace MQTTnet.Core.Client
namespace MQTTnet.Core
{
public interface IApplicationMessageReceiver
{
MQTTnet.Core/ManagedClient/ManagedMqttClient.cs
+3 -3
View File
@@ -91,16 +91,15 @@ namespace MQTTnet.Core.ManagedClient
return Task.FromResult(0);
}
public Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
public async Task PublishAsync(IEnumerable<MqttApplicationMessage> 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<TopicFilter> topicFilters)
@@ -208,6 +207,7 @@ namespace MQTTnet.Core.ManagedClient
}
await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
await _storageManager.RemoveAsync(message).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
MQTTnet.Core/ManagedClient/ManagedMqttClientExtensions.cs
+2 -10
View File
@@ -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));
MQTTnet.Core/Client/MqttApplicationMessageReceivedEventArgs.cs → MQTTnet.Core/MqttApplicationMessageReceivedEventArgs.cs
+1 -1
View File
@@ -1,6 +1,6 @@
using System;
namespace MQTTnet.Core.Client
namespace MQTTnet.Core
{
public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
{
MQTTnet.Core/Server/IMqttServer.cs
+1
View File
@@ -11,6 +11,7 @@ namespace MQTTnet.Core.Server
event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
IList<ConnectedMqttClient> GetConnectedClients();
void Publish(IEnumerable<MqttApplicationMessage> applicationMessages);
Task StartAsync();
Task StopAsync();
MQTTnet.Core/Server/IMqttServerFactory.cs
+5 -1
View File
@@ -1,7 +1,11 @@
namespace MQTTnet.Core.Server
using System;
namespace MQTTnet.Core.Server
{
public interface IMqttServerFactory
{
IMqttServer CreateMqttServer();
IMqttServer CreateMqttServer(Action<MqttServerOptions> configure);
}
}
MQTTnet.Core/Server/MqttApplicationMessageInterceptorContext.cs
+7
View File
@@ -0,0 +1,7 @@
namespace MQTTnet.Core.Server
{
public class MqttApplicationMessageInterceptorContext
{
public MqttApplicationMessage ApplicationMessage { get; set; }
}
}
MQTTnet.Core/Server/MqttClientSession.cs
+76 -39
View File
@@ -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<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
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<MqttClientSession> _logger;
@@ -26,14 +27,22 @@ namespace MQTTnet.Core.Server
private CancellationTokenSource _cancellationTokenSource;
private MqttApplicationMessage _willMessage;
public MqttClientSession(string clientId, MqttClientSessionsManager mqttClientSessionsManager, ILogger<MqttClientSession> logger, ILogger<MqttClientPendingMessagesQueue> msgQueueLogger)
public MqttClientSession(
string clientId,
IOptions<MqttServerOptions> options,
MqttClientSessionsManager sessionsManager,
MqttClientSubscriptionsManager subscriptionsManager,
ILogger<MqttClientSession> logger,
ILogger<MqttClientPendingMessagesQueue> 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<MqttPubRelPacket>());
}
else if (packet is MqttPubRecPacket pubRecPacket)
{
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, pubRecPacket.CreateResponse<MqttPubRelPacket>());
}
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 });
}
}
}
MQTTnet.Core/Server/MqttClientSessionsManager.cs
+18 -15
View File
@@ -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<string, MqttClientSession> _clientSessions = new Dictionary<string, MqttClientSession>();
private readonly ILogger<MqttClientSessionsManager> _logger;
private readonly IMqttClientSesssionFactory _mqttClientSesssionFactory;
private readonly IMqttClientSesssionFactory _clientSesssionFactory;
private readonly MqttServerOptions _options;
public MqttClientSessionsManager(IOptions<MqttServerOptions> options, ILogger<MqttClientSessionsManager> logger, MqttClientRetainedMessagesManager retainedMessagesManager, IMqttClientSesssionFactory mqttClientSesssionFactory)
public MqttClientSessionsManager(
IOptions<MqttServerOptions> options,
ILogger<MqttClientSessionsManager> 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<MqttClientConnectedEventArgs> ClientConnected;
@@ -34,14 +38,13 @@ namespace MQTTnet.Core.Server
public event EventHandler<MqttApplicationMessageReceivedEventArgs> 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);
MQTTnet.Core/Server/MqttClientSubscribeResult.cs
+11
View File
@@ -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; }
}
}
MQTTnet.Core/Server/MqttClientSubscriptionsManager.cs
+27 -4
View File
@@ -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<string, MqttQualityOfServiceLevel> _subscribedTopics = new Dictionary<string, MqttQualityOfServiceLevel>();
private readonly MqttServerOptions _options;
public MqttSubAckPacket Subscribe(MqttSubscribePacket subscribePacket)
public MqttClientSubscriptionsManager(IOptions<MqttServerOptions> 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<MqttSubAckPacket>();
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)
MQTTnet.Core/Server/MqttServer.cs
+15 -6
View File
@@ -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<MqttClientDisconnectedEventArgs> ClientDisconnected;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
public void Publish(IEnumerable<MqttApplicationMessage> 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<MqttApplicationMessage> 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)
MQTTnet.Core/Server/MqttServerOptions.cs
+3 -1
View File
@@ -16,7 +16,9 @@ namespace MQTTnet.Core.Server
public Func<MqttConnectPacket, MqttConnectReturnCode> ConnectionValidator { get; set; }
public Func<MqttApplicationMessage, MqttApplicationMessage> ApplicationMessageInterceptor { get; set; }
public Action<MqttApplicationMessageInterceptorContext> ApplicationMessageInterceptor { get; set; }
public Action<MqttSubscriptionInterceptorContext> SubscriptionsInterceptor { get; set; }
public IMqttServerStorage Storage { get; set; }
}
MQTTnet.Core/Server/MqttSubscribeResult.cs
+11
View File
@@ -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; }
}
}
MQTTnet.Core/Server/MqttSubscriptionInterceptorContext.cs
+19
View File
@@ -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; }
}
}
MQTTnet.sln
+4 -21
View File
@@ -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}
Tests/MQTTnet.Core.Tests/MqttServerTests.cs
+205 -117
View File
@@ -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<IMqttServerAdapter>(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<IMqttServerAdapter>(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<IMqttServerAdapter>(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<IMqttServerAdapter>(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<IMqttServerAdapter>(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<IMqttServerAdapter>(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<IMqttServerAdapter>(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<IMqttServer>();
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<IMqttServerAdapter>(serverAdapter)
.BuildServiceProvider();
var s = services.GetRequiredService<IMqttServer>();
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<IMqttServerAdapter>(serverAdapter)
.BuildServiceProvider();
var s = services.GetRequiredService<IMqttServer>();
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);
}
}
Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs
+8 -7
View File
@@ -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<MqttServerOptions>(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<MqttServerOptions>(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<MqttServerOptions>(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
{
Tests/MQTTnet.Core.Tests/TestClientSessionFactory.cs
+5 -3
View File
@@ -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<MqttClientSession>(), new TestLogger<MqttClientPendingMessagesQueue>());
throw new NotImplementedException();
//return new MqttClientSession(clientId, mqttClientSessionsManager, new TestLogger<MqttClientSession>(), new TestLogger<MqttClientPendingMessagesQueue>());
}
}
}
Tests/MQTTnet.Core.Tests/TestLogger.cs
+6 -2
View File
@@ -3,11 +3,11 @@ using System;
namespace MQTTnet.Core.Tests
{
public class TestLogger<T> : ILogger<T>
public class TestLogger<T> : IDisposable, ILogger<T>
{
public IDisposable BeginScope<TState>(TState state)
{
throw new NotImplementedException();
return this;
}
public bool IsEnabled(LogLevel logLevel)
@@ -18,5 +18,9 @@ namespace MQTTnet.Core.Tests
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func<TState, Exception, string> formatter)
{
}
public void Dispose()
{
}
}
}
Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj
+1 -1
View File
@@ -14,7 +14,7 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Frameworks\MQTTnet.AspnetCore\MQTTnet.AspnetCore.csproj" />
<ProjectReference Include="..\..\Frameworks\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj" />
</ItemGroup>
<ItemGroup>
Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs
+1 -1
View File
@@ -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;
Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs
+3 -3
View File
@@ -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();
Tests/MQTTnet.TestApp.NetCore/ServerTest.cs
+4 -6
View File
@@ -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"));
}
};
});
Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj
+4 -4
View File
@@ -18,7 +18,7 @@
<ProjectTypeGuids>{A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids>
<WindowsXamlEnableOverview>true</WindowsXamlEnableOverview>
<PackageCertificateKeyFile>MQTTnet.TestApp.UniversalWindows_TemporaryKey.pfx</PackageCertificateKeyFile>
<PackageCertificateThumbprint>13E377A693C923EE9E88EE84B32A4B9881657362</PackageCertificateThumbprint>
<PackageCertificateThumbprint>FA56408F253A1F7FCAC649611411B6B9F42BC4F9</PackageCertificateThumbprint>
<RuntimeIdentifiers>win10;win10-arm;win10-arm-aot;win10-x86;win10-x86-aot;win10-x64;win10-x64-aot</RuntimeIdentifiers>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Debug|x86'">
@@ -126,9 +126,9 @@
</Page>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Frameworks\MQTTnet.UniversalWindows\MQTTnet.UniversalWindows.csproj">
<Project>{bd60c727-d8e8-40c3-b8e3-c95a864ae611}</Project>
<Name>MQTTnet.UniversalWindows</Name>
<ProjectReference Include="..\..\Frameworks\MQTTnet.Netstandard\MQTTnet.NetStandard.csproj">
<Project>{3587e506-55a2-4eb3-99c7-dc01e42d25d2}</Project>
<Name>MQTTnet.NetStandard</Name>
</ProjectReference>
<ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj">
<Project>{2ecb99e4-72d0-4c23-99ba-93d511d3967d}</Project>
Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows_TemporaryKey.pfx
BIN
View File
Binary file not shown.
Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml
+11 -1
View File
@@ -65,7 +65,6 @@
<Button Click="Publish" Width="120">Publish</Button>
</StackPanel>
</PivotItem>
<PivotItem Header="Subscribe">
<StackPanel Background="{ThemeResource ApplicationPageBackgroundThemeBrush}">
<TextBlock>Topic:</TextBlock>
@@ -84,6 +83,17 @@
</StackPanel>
</StackPanel>
</PivotItem>
<PivotItem Header="Server">
<StackPanel Background="{ThemeResource ApplicationPageBackgroundThemeBrush}">
<TextBlock>Port:</TextBlock>
<TextBox x:Name="ServerPort" Text="1883"></TextBox>
<StackPanel Orientation="Horizontal">
<Button Width="120" Margin="0,0,10,0" Click="StartServer">Start</Button>
<Button Width="120" Margin="0,0,10,0" Click="StopServer">Stop</Button>
</StackPanel>
</StackPanel>
</PivotItem>
</Pivot>
<ScrollViewer Margin="12,10,12,0" Grid.Row="1" Background="Black" Foreground="WhiteSmoke" FontFamily="Consolas" HorizontalScrollMode="Enabled" VerticalScrollMode="Enabled" HorizontalScrollBarVisibility="Visible" VerticalScrollBarVisibility="Visible">
Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs
+28
View File
@@ -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;
}
}
}
Tests/MQTTnet.TestApp.UniversalWindows/Package.appxmanifest
+1 -1
View File
@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="utf-8"?>
<Package xmlns="http://schemas.microsoft.com/appx/manifest/foundation/windows10" xmlns:mp="http://schemas.microsoft.com/appx/2014/phone/manifest" xmlns:uap="http://schemas.microsoft.com/appx/manifest/uap/windows10" xmlns:uap3="http://schemas.microsoft.com/appx/manifest/uap/windows10/3" IgnorableNamespaces="uap mp uap3">
<Identity Name="4fa21172-9128-4e84-8a6d-74b9acde4d58" Publisher="CN=Christian" Version="1.0.0.0" />
<Identity Name="4fa21172-9128-4e84-8a6d-74b9acde4d58" Publisher="CN=MQTT_DEV" Version="1.0.0.0" />
<mp:PhoneIdentity PhoneProductId="4fa21172-9128-4e84-8a6d-74b9acde4d58" PhonePublisherId="00000000-0000-0000-0000-000000000000" />
<Properties>
<DisplayName>MQTTnet.TestApp.UniversalWindows</DisplayName>
appveyor.yml
+15
View File
@@ -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