YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-20 19:58:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' of https://github.com/chkr1011/MQTTnet into develop

Browse Source
This commit is contained in:
Christian Kratky
2017-10-28 10:34:32 +02:00
parent ba16ae6568 3b2c40d49e
commit d560cb4c9e
17 changed files with 276 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Tests/MQTTnet.Core.Tests/MqttServerTests.cs
+203 -114
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);
}
@@ -253,30 +329,37 @@ namespace MQTTnet.Core.Tests
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
@@ -305,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/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;