YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-11-04 21:43:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

improved api surface of mqttserver and fixed unittests

Browse Source
This commit is contained in:
JanEggers
2017-10-02 22:55:56 +02:00
parent 3079ad37ce
commit 1d78d675f7
9 changed files with 90 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MQTTnet.Core/Server/IMqttServer.cs
+1 -2
View File
@@ -11,8 +11,7 @@ namespace MQTTnet.Core.Server
event EventHandler<MqttClientConnectedEventArgs> ClientConnected; event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected; event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
IList<ConnectedMqttClient> GetConnectedClients(); IReadOnlyList<ConnectedMqttClient> GetConnectedClients();
void InjectClient(IMqttCommunicationAdapter adapter);
void Publish(MqttApplicationMessage applicationMessage); void Publish(MqttApplicationMessage applicationMessage);
Task StartAsync(); Task StartAsync();
MQTTnet.Core/Server/MqttClientSession.cs
+2 -2
View File
@@ -157,12 +157,12 @@ namespace MQTTnet.Core.Server
} }
else if (packet is MqttDisconnectPacket || packet is MqttConnectPacket) else if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
{ {
_cancellationTokenSource.Cancel(); Stop();
} }
else else
{ {
MqttTrace.Warning(nameof(MqttClientSession), "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); MqttTrace.Warning(nameof(MqttClientSession), "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
_cancellationTokenSource.Cancel(); Stop();
} }
} }
MQTTnet.Core/Server/MqttClientSessionsManager.cs
+1 -1
View File
@@ -104,7 +104,7 @@ namespace MQTTnet.Core.Server
} }
} }
public IList<ConnectedMqttClient> GetConnectedClients() public IReadOnlyList<ConnectedMqttClient> GetConnectedClients()
{ {
lock (_clientSessions) lock (_clientSessions)
{ {
MQTTnet.Core/Server/MqttServer.cs
+1 -9
View File
@@ -27,7 +27,7 @@ namespace MQTTnet.Core.Server
_clientSessionsManager.ClientDisconnected += OnClientDisconnected; _clientSessionsManager.ClientDisconnected += OnClientDisconnected;
} }
public IList<ConnectedMqttClient> GetConnectedClients() public IReadOnlyList<ConnectedMqttClient> GetConnectedClients()
{ {
return _clientSessionsManager.GetConnectedClients(); return _clientSessionsManager.GetConnectedClients();
} }
@@ -43,14 +43,6 @@ namespace MQTTnet.Core.Server
_clientSessionsManager.DispatchPublishPacket(null, applicationMessage.ToPublishPacket()); _clientSessionsManager.DispatchPublishPacket(null, applicationMessage.ToPublishPacket());
} }
public void InjectClient(IMqttCommunicationAdapter adapter)
{
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
if (_cancellationTokenSource == null) throw new InvalidOperationException("The MQTT server is not started.");
OnClientAccepted(adapter);
}
public async Task StartAsync() public async Task StartAsync()
{ {
if (_cancellationTokenSource != null) throw new InvalidOperationException("The MQTT server is already started."); if (_cancellationTokenSource != null) throw new InvalidOperationException("The MQTT server is already started.");
MQTTnet.sln
+8 -8
View File
@@ -50,18 +50,18 @@ Global
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|Any CPU.Build.0 = Debug|Any CPU {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|Any CPU.Build.0 = Debug|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|ARM.ActiveCfg = Debug|Any CPU {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|ARM.ActiveCfg = Debug|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|ARM.Build.0 = Debug|Any CPU {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|ARM.Build.0 = Debug|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x64.ActiveCfg = Debug|x64 {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x64.ActiveCfg = Debug|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x64.Build.0 = Debug|x64 {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x64.Build.0 = Debug|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x86.ActiveCfg = Debug|x86 {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x86.ActiveCfg = Debug|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x86.Build.0 = Debug|x86 {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x86.Build.0 = Debug|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|Any CPU.ActiveCfg = Release|Any CPU {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|Any CPU.ActiveCfg = Release|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|Any CPU.Build.0 = Release|Any CPU {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|Any CPU.Build.0 = Release|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|ARM.ActiveCfg = Release|Any CPU {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|ARM.ActiveCfg = Release|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|ARM.Build.0 = Release|Any CPU {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|ARM.Build.0 = Release|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x64.ActiveCfg = Release|x64 {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x64.ActiveCfg = Release|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x64.Build.0 = Release|x64 {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x64.Build.0 = Release|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.ActiveCfg = Release|x86 {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.ActiveCfg = Release|Any CPU
{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.Build.0 = Release|x86 {A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.Build.0 = Release|Any CPU
{A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Debug|Any CPU.Build.0 = Debug|Any CPU {A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Debug|Any CPU.Build.0 = Debug|Any CPU
{A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Debug|ARM.ActiveCfg = Debug|Any CPU {A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Debug|ARM.ActiveCfg = Debug|Any CPU
Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj
+13 -108
View File
@@ -1,115 +1,20 @@
<?xml version="1.0" encoding="utf-8"?> <Project Sdk="Microsoft.NET.Sdk">
<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> <PropertyGroup>
<Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration> <OutputType>Exe</OutputType>
<Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform> <DebugType>Full</DebugType>
<ProjectGuid>{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}</ProjectGuid> <TargetFramework>netcoreapp2.0</TargetFramework>
<OutputType>Library</OutputType>
<AppDesignerFolder>Properties</AppDesignerFolder>
<RootNamespace>MQTTnet.Core.Tests</RootNamespace>
<AssemblyName>MQTTnet.Core.Tests</AssemblyName>
<TargetFrameworkVersion>v4.5.2</TargetFrameworkVersion>
<FileAlignment>512</FileAlignment>
<TargetFrameworkProfile />
</PropertyGroup>
<PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
<DebugSymbols>true</DebugSymbols>
<DebugType>full</DebugType>
<Optimize>false</Optimize>
<OutputPath>bin\Debug\</OutputPath>
<DefineConstants>DEBUG;TRACE</DefineConstants>
<ErrorReport>prompt</ErrorReport>
<WarningLevel>4</WarningLevel>
</PropertyGroup>
<PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
<DebugType>pdbonly</DebugType>
<Optimize>true</Optimize>
<OutputPath>bin\Release\</OutputPath>
<DefineConstants>TRACE</DefineConstants>
<ErrorReport>prompt</ErrorReport>
<WarningLevel>4</WarningLevel>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Any CPU|AnyCPU'">
<OutputPath>bin\Any CPU\</OutputPath>
<DefineConstants>TRACE</DefineConstants>
<Optimize>true</Optimize>
<DebugType>pdbonly</DebugType>
<PlatformTarget>AnyCPU</PlatformTarget>
<ErrorReport>prompt</ErrorReport>
<CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Debug|x64'">
<DebugSymbols>true</DebugSymbols>
<OutputPath>bin\x64\Debug\</OutputPath>
<DefineConstants>DEBUG;TRACE</DefineConstants>
<DebugType>full</DebugType>
<PlatformTarget>x64</PlatformTarget>
<ErrorReport>prompt</ErrorReport>
<CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Release|x64'">
<OutputPath>bin\x64\Release\</OutputPath>
<DefineConstants>TRACE</DefineConstants>
<Optimize>true</Optimize>
<DebugType>pdbonly</DebugType>
<PlatformTarget>x64</PlatformTarget>
<ErrorReport>prompt</ErrorReport>
<CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Any CPU|x64'">
<OutputPath>bin\x64\Any CPU\</OutputPath>
<DefineConstants>TRACE</DefineConstants>
<Optimize>true</Optimize>
<DebugType>pdbonly</DebugType>
<PlatformTarget>x64</PlatformTarget>
<ErrorReport>prompt</ErrorReport>
<CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Debug|x86'">
<PlatformTarget>x86</PlatformTarget>
<OutputPath>bin\x86\Debug\</OutputPath>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Release|x86'">
<PlatformTarget>x86</PlatformTarget>
<OutputPath>bin\x86\Release\</OutputPath>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Any CPU|x86'">
<PlatformTarget>x86</PlatformTarget>
<OutputPath>bin\x86\Any CPU\</OutputPath>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
<Reference Include="Microsoft.VisualStudio.QualityTools.UnitTestFramework, Version=10.1.0.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL" /> <PackageReference Include="MSTest.TestAdapter" Version="1.1.18" />
<Reference Include="System" /> <PackageReference Include="MSTest.TestFramework" Version="1.1.18" />
<Reference Include="System.Core" /> <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.3.0" />
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<Compile Include="ByteReaderTests.cs" /> <ProjectReference Include="..\..\Frameworks\MQTTnet.Netstandard\MQTTnet.NetStandard.csproj" />
<Compile Include="ByteWriterTests.cs" /> <ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj" />
<Compile Include="ExtensionTests.cs" />
<Compile Include="MqttPacketSerializerTests.cs" />
<Compile Include="MqttServerTests.cs" />
<Compile Include="MqttSubscriptionsManagerTests.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
<Compile Include="TestMqttCommunicationAdapter.cs" />
<Compile Include="TestMqttServerAdapter.cs" />
<Compile Include="TopicFilterComparerTests.cs" />
</ItemGroup> </ItemGroup>
<ItemGroup>
<Service Include="{82A7F48D-3B50-4B1E-B82E-3ADA8210C358}" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj">
<Project>{2ecb99e4-72d0-4c23-99ba-93d511d3967d}</Project>
<Name>MQTTnet.Core</Name>
</ProjectReference>
</ItemGroup>
<Import Project="$(MSBuildToolsPath)\Microsoft.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> </Project>
Tests/MQTTnet.Core.Tests/MqttServerTests.cs
+24 -32
View File
@@ -6,6 +6,7 @@ using MQTTnet.Core.Client;
using MQTTnet.Core.Packets; using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol; using MQTTnet.Core.Protocol;
using MQTTnet.Core.Server; using MQTTnet.Core.Server;
using System;
namespace MQTTnet.Core.Tests namespace MQTTnet.Core.Tests
{ {
@@ -48,12 +49,13 @@ namespace MQTTnet.Core.Tests
[TestMethod] [TestMethod]
public async Task MqttServer_WillMessage() public async Task MqttServer_WillMessage()
{ {
var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { new TestMqttServerAdapter() }); var serverAdapter = new TestMqttServerAdapter();
s.StartAsync(); var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
await s.StartAsync();
var willMessage = new MqttApplicationMessage("My/last/will", new byte[0], MqttQualityOfServiceLevel.AtMostOnce, false); var willMessage = new MqttApplicationMessage("My/last/will", new byte[0], MqttQualityOfServiceLevel.AtMostOnce, false);
var c1 = ConnectTestClient("c1", null, s); var c1 = await serverAdapter.ConnectTestClient(s, "c1");
var c2 = ConnectTestClient("c2", willMessage, s); var c2 = await serverAdapter.ConnectTestClient(s, "c2", willMessage);
var receivedMessagesCount = 0; var receivedMessagesCount = 0;
c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
@@ -63,7 +65,7 @@ namespace MQTTnet.Core.Tests
await Task.Delay(1000); await Task.Delay(1000);
s.StopAsync(); await s.StopAsync();
Assert.AreEqual(1, receivedMessagesCount); Assert.AreEqual(1, receivedMessagesCount);
} }
@@ -71,11 +73,12 @@ namespace MQTTnet.Core.Tests
[TestMethod] [TestMethod]
public async Task MqttServer_Unsubscribe() public async Task MqttServer_Unsubscribe()
{ {
var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { new TestMqttServerAdapter() }); var serverAdapter = new TestMqttServerAdapter();
s.StartAsync(); var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
await s.StartAsync();
var c1 = ConnectTestClient("c1", null, s); var c1 = await serverAdapter.ConnectTestClient(s, "c1");
var c2 = ConnectTestClient("c2", null, s); var c2 = await serverAdapter.ConnectTestClient(s, "c2");
var receivedMessagesCount = 0; var receivedMessagesCount = 0;
c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
@@ -97,7 +100,7 @@ namespace MQTTnet.Core.Tests
await Task.Delay(500); await Task.Delay(500);
Assert.AreEqual(1, receivedMessagesCount); Assert.AreEqual(1, receivedMessagesCount);
s.StopAsync(); await s.StopAsync();
await Task.Delay(500); await Task.Delay(500);
Assert.AreEqual(1, receivedMessagesCount); Assert.AreEqual(1, receivedMessagesCount);
@@ -106,10 +109,11 @@ namespace MQTTnet.Core.Tests
[TestMethod] [TestMethod]
public async Task MqttServer_Publish() public async Task MqttServer_Publish()
{ {
var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { new TestMqttServerAdapter() }); var serverAdapter = new TestMqttServerAdapter();
s.StartAsync(); var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
await s.StartAsync();
var c1 = ConnectTestClient("c1", null, s); var c1 = await serverAdapter.ConnectTestClient(s, "c1");
var receivedMessagesCount = 0; var receivedMessagesCount = 0;
c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
@@ -120,24 +124,11 @@ namespace MQTTnet.Core.Tests
s.Publish(message); s.Publish(message);
await Task.Delay(500); await Task.Delay(500);
s.StopAsync(); await s.StopAsync();
Assert.AreEqual(1, receivedMessagesCount); Assert.AreEqual(1, receivedMessagesCount);
} }
private static MqttClient ConnectTestClient(string clientId, MqttApplicationMessage willMessage, MqttServer server)
{
var adapterA = new TestMqttCommunicationAdapter();
var adapterB = new TestMqttCommunicationAdapter();
adapterA.Partner = adapterB;
adapterB.Partner = adapterA;
var client = new MqttClient(new MqttClientOptions(), adapterA);
server.InjectClient(clientId, adapterB);
client.ConnectAsync(willMessage).Wait();
return client;
}
private async Task TestPublishAsync( private async Task TestPublishAsync(
string topic, string topic,
MqttQualityOfServiceLevel qualityOfServiceLevel, MqttQualityOfServiceLevel qualityOfServiceLevel,
@@ -145,11 +136,12 @@ namespace MQTTnet.Core.Tests
MqttQualityOfServiceLevel filterQualityOfServiceLevel, MqttQualityOfServiceLevel filterQualityOfServiceLevel,
int expectedReceivedMessagesCount) int expectedReceivedMessagesCount)
{ {
var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { new TestMqttServerAdapter() }); var serverAdapter = new TestMqttServerAdapter();
s.StartAsync(); var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
await s.StartAsync();
var c1 = ConnectTestClient("c1", null, s); var c1 = await serverAdapter.ConnectTestClient(s, "c1");
var c2 = ConnectTestClient("c2", null, s); var c2 = await serverAdapter.ConnectTestClient(s, "c2");
var receivedMessagesCount = 0; var receivedMessagesCount = 0;
c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
@@ -162,7 +154,7 @@ namespace MQTTnet.Core.Tests
await Task.Delay(500); await Task.Delay(500);
s.StopAsync(); await s.StopAsync();
Assert.AreEqual(expectedReceivedMessagesCount, receivedMessagesCount); Assert.AreEqual(expectedReceivedMessagesCount, receivedMessagesCount);
} }
Tests/MQTTnet.Core.Tests/Properties/AssemblyInfo.cs
-15
View File
@@ -1,15 +0,0 @@
using System.Reflection;
using System.Runtime.InteropServices;
[assembly: AssemblyTitle("MQTTnet.Core.Tests")]
[assembly: AssemblyDescription("")]
[assembly: AssemblyConfiguration("")]
[assembly: AssemblyCompany("Christian Kratky")]
[assembly: AssemblyProduct("MQTTnet")]
[assembly: AssemblyCopyright("Copyright © Christian Kratky 2016-2017")]
[assembly: AssemblyTrademark("")]
[assembly: AssemblyCulture("")]
[assembly: ComVisible(false)]
[assembly: Guid("a7ff0c91-25de-4ba6-b39e-f54e8dadf1cc")]
[assembly: AssemblyVersion("1.0.0.0")]
[assembly: AssemblyFileVersion("1.0.0.0")]
Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs
+39 -3
View File
@@ -2,16 +2,52 @@
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Core.Adapter; using MQTTnet.Core.Adapter;
using MQTTnet.Core.Server; using MQTTnet.Core.Server;
using MQTTnet.Core.Client;
namespace MQTTnet.Core.Tests namespace MQTTnet.Core.Tests
{ {
public class TestMqttServerAdapter : IMqttServerAdapter public class TestMqttServerAdapter : IMqttServerAdapter
{ {
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected; public event Action<IMqttCommunicationAdapter> ClientAccepted;
public void FireClientConnectedEvent(MqttClientConnectedEventArgs eventArgs) public async Task<MqttClient> ConnectTestClient(IMqttServer server, string clientId, MqttApplicationMessage willMessage = null)
{ {
ClientConnected?.Invoke(this, eventArgs); var adapterA = new TestMqttCommunicationAdapter();
var adapterB = new TestMqttCommunicationAdapter();
adapterA.Partner = adapterB;
adapterB.Partner = adapterA;
var client = new MqttClient(new MqttClientOptions() { ClientId = clientId }, adapterA);
var connected = WaitForClientToConnect(server, clientId);
FireClientAcceptedEvent(adapterB);
await client.ConnectAsync(willMessage);
await connected;
return client;
}
private static Task WaitForClientToConnect(IMqttServer s, string clientId)
{
var tcs = new TaskCompletionSource<object>();
EventHandler<MqttClientConnectedEventArgs> handler = null;
handler = (sender, args) =>
{
if (args.Client.ClientId == clientId)
{
s.ClientConnected -= handler;
tcs.SetResult(null);
}
};
s.ClientConnected += handler;
return tcs.Task;
}
private void FireClientAcceptedEvent(IMqttCommunicationAdapter adapter)
{
ClientAccepted?.Invoke(adapter);
} }
public Task StartAsync(MqttServerOptions options) public Task StartAsync(MqttServerOptions options)