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

Merge branch 'master' into feature/low_level_client

Browse Source
This commit is contained in:
Christian Kratky
2020-04-01 23:01:03 +02:00
parent 04ccdcd562 af0b98eace
commit 0ca4c88e69
35 changed files with 373 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs
+3 -3
View File
@@ -1,11 +1,11 @@
using BenchmarkDotNet.Attributes;
using MQTTnet.Channel;
using MQTTnet.Client.Options;
using MQTTnet.Diagnostics;
using MQTTnet.Implementations;
using MQTTnet.Server;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client.Options;
namespace MQTTnet.Benchmarks
{
@@ -20,7 +20,7 @@ namespace MQTTnet.Benchmarks
public void Setup()
{
var factory = new MqttFactory();
var tcpServer = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger());
var tcpServer = new MqttTcpServerAdapter(new MqttNetLogger());
tcpServer.ClientHandler += args =>
{
_serverChannel =
@@ -30,7 +30,7 @@ namespace MQTTnet.Benchmarks
return Task.CompletedTask;
};
_mqttServer = factory.CreateMqttServer(new[] { tcpServer }, new MqttNetLogger());
var serverOptions = new MqttServerOptionsBuilder().Build();
Tests/MQTTnet.Core.Tests/ManagedMqttClient_Tests.cs
+12 -10
View File
@@ -1,9 +1,4 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Options;
@@ -12,6 +7,11 @@ using MQTTnet.Diagnostics;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Server;
using MQTTnet.Tests.Mockups;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.Tests
{
@@ -69,7 +69,7 @@ namespace MQTTnet.Tests
.WithTcpServer("localhost", testEnvironment.ServerPort)
.WithWillMessage(willMessage);
var dyingClient = testEnvironment.CreateClient();
var dyingManagedClient = new ManagedMqttClient(dyingClient, testEnvironment.ClientLogger.CreateChildLogger());
var dyingManagedClient = new ManagedMqttClient(dyingClient, testEnvironment.ClientLogger);
await dyingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder()
.WithClientOptions(clientOptions)
.Build());
@@ -96,7 +96,7 @@ namespace MQTTnet.Tests
var server = await testEnvironment.StartServerAsync();
var managedClient = new ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetLogger().CreateChildLogger());
var managedClient = new ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetLogger());
var clientOptions = new MqttClientOptionsBuilder()
.WithTcpServer("localhost", testEnvironment.ServerPort);
@@ -110,6 +110,8 @@ namespace MQTTnet.Tests
await managedClient.StopAsync();
await Task.Delay(500);
Assert.AreEqual(0, (await server.GetClientStatusAsync()).Count);
}
}
@@ -126,7 +128,7 @@ namespace MQTTnet.Tests
var server = await testEnvironment.StartServerAsync();
var managedClient = new ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetLogger().CreateChildLogger());
var managedClient = new ManagedMqttClient(testEnvironment.CreateClient(), new MqttNetLogger());
var clientOptions = new MqttClientOptionsBuilder()
.WithTcpServer("localhost", testEnvironment.ServerPort);
var storage = new ManagedMqttClientTestStorage();
@@ -349,7 +351,7 @@ namespace MQTTnet.Tests
managedOptions.ConnectionCheckInterval = connectionCheckInterval ?? TimeSpan.FromSeconds(0.1);
var managedClient =
new ManagedMqttClient(underlyingClient ?? testEnvironment.CreateClient(), new MqttNetLogger().CreateChildLogger());
new ManagedMqttClient(underlyingClient ?? testEnvironment.CreateClient(), new MqttNetLogger());
var connected = GetConnectedTask(managedClient);
Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs
+22 -25
View File
@@ -29,37 +29,34 @@ namespace MQTTnet.Tests.Mockups
public IMqttClientOptions Options => Implementation.Options;
public IMqttClientConnectedHandler ConnectedHandler { get => Implementation.ConnectedHandler; set => Implementation.ConnectedHandler = value; }
public IMqttClientDisconnectedHandler DisconnectedHandler { get => Implementation.DisconnectedHandler; set => Implementation.DisconnectedHandler = value; }
public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get => Implementation.ApplicationMessageReceivedHandler; set => Implementation.ApplicationMessageReceivedHandler = value; }
public IMqttClientConnectedHandler ConnectedHandler
{
get => Implementation.ConnectedHandler;
set => Implementation.ConnectedHandler = value;
}
public IMqttClientDisconnectedHandler DisconnectedHandler
{
get => Implementation.DisconnectedHandler;
set => Implementation.DisconnectedHandler = value;
}
public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler
{
get => Implementation.ApplicationMessageReceivedHandler;
set => Implementation.ApplicationMessageReceivedHandler = value;
}
public Task<MqttClientAuthenticateResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken)
{
if (TestContext != null)
{
switch (options)
var clientOptions = (MqttClientOptions)options;
var existingClientId = clientOptions.ClientId;
if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName))
{
case MqttClientOptionsBuilder builder:
{
var existingClientId = builder.Build().ClientId;
if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName))
{
builder.WithClientId(TestContext.TestName + existingClientId);
}
break;
}
case MqttClientOptions op:
{
var existingClientId = op.ClientId;
if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName))
{
op.ClientId = TestContext.TestName + existingClientId;
}
break;
}
clientOptions.ClientId = TestContext.TestName + existingClientId;
}
}
Tests/MQTTnet.Core.Tests/Mockups/TestLogger.cs
+1 -1
View File
@@ -7,7 +7,7 @@ namespace MQTTnet.Tests.Mockups
{
public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
public IMqttNetLogger CreateChildLogger(string source = null)
public IMqttNetLogger CreateChildLogger(string source)
{
return new TestLogger();
}
Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs
+46 -27
View File
@@ -22,16 +22,50 @@ namespace MQTTnet.Tests.Mockups
public IMqttServer Implementation { get; }
public TestContext TestContext { get; }
public TestEnvironment TestEnvironment { get; }
public IMqttServerStartedHandler StartedHandler { get => Implementation.StartedHandler; set => Implementation.StartedHandler = value; }
public IMqttServerStoppedHandler StoppedHandler { get => Implementation.StoppedHandler; set => Implementation.StoppedHandler = value; }
public IMqttServerClientConnectedHandler ClientConnectedHandler { get => Implementation.ClientConnectedHandler; set => Implementation.ClientConnectedHandler = value; }
public IMqttServerClientDisconnectedHandler ClientDisconnectedHandler { get => Implementation.ClientDisconnectedHandler; set => Implementation.ClientDisconnectedHandler = value; }
public IMqttServerClientSubscribedTopicHandler ClientSubscribedTopicHandler { get => Implementation.ClientSubscribedTopicHandler; set => Implementation.ClientSubscribedTopicHandler = value; }
public IMqttServerClientUnsubscribedTopicHandler ClientUnsubscribedTopicHandler { get => Implementation.ClientUnsubscribedTopicHandler; set => Implementation.ClientUnsubscribedTopicHandler = value; }
public IMqttServerStartedHandler StartedHandler
{
get => Implementation.StartedHandler;
set => Implementation.StartedHandler = value;
}
public IMqttServerStoppedHandler StoppedHandler
{
get => Implementation.StoppedHandler;
set => Implementation.StoppedHandler = value;
}
public IMqttServerClientConnectedHandler ClientConnectedHandler
{
get => Implementation.ClientConnectedHandler;
set => Implementation.ClientConnectedHandler = value;
}
public IMqttServerClientDisconnectedHandler ClientDisconnectedHandler
{
get => Implementation.ClientDisconnectedHandler;
set => Implementation.ClientDisconnectedHandler = value;
}
public IMqttServerClientSubscribedTopicHandler ClientSubscribedTopicHandler
{
get => Implementation.ClientSubscribedTopicHandler;
set => Implementation.ClientSubscribedTopicHandler = value;
}
public IMqttServerClientUnsubscribedTopicHandler ClientUnsubscribedTopicHandler
{
get => Implementation.ClientUnsubscribedTopicHandler;
set => Implementation.ClientUnsubscribedTopicHandler = value;
}
public IMqttServerOptions Options => Implementation.Options;
public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get => Implementation.ApplicationMessageReceivedHandler; set => Implementation.ApplicationMessageReceivedHandler = value; }
public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler
{
get => Implementation.ApplicationMessageReceivedHandler;
set => Implementation.ApplicationMessageReceivedHandler = value;
}
public Task ClearRetainedApplicationMessagesAsync()
{
@@ -62,26 +96,11 @@ namespace MQTTnet.Tests.Mockups
{
if (TestContext != null)
{
switch (options)
var serverOptions = (MqttServerOptions)options;
if (serverOptions.ConnectionValidator == null)
{
case MqttServerOptionsBuilder builder:
{
if (builder.Build().ConnectionValidator == null)
{
builder.WithConnectionValidator(ConnectionValidator);
}
break;
}
case MqttServerOptions op:
{
if (op.ConnectionValidator == null)
{
op.ConnectionValidator = new MqttServerConnectionValidatorDelegate(ConnectionValidator);
}
break;
}
serverOptions.ConnectionValidator = new MqttServerConnectionValidatorDelegate(ConnectionValidator);
}
}
@@ -92,7 +111,7 @@ namespace MQTTnet.Tests.Mockups
{
if (!ctx.ClientId.StartsWith(TestContext.TestName))
{
TestEnvironment.TrackException(new InvalidOperationException($"invalid client connected '{ctx.ClientId}'"));
TestEnvironment.TrackException(new InvalidOperationException($"Invalid client ID used ({ctx.ClientId}). It must start with UnitTest name."));
ctx.ReasonCode = Protocol.MqttConnectReasonCode.ClientIdentifierNotValid;
}
}
Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs
+15 -13
View File
@@ -1,9 +1,3 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Client.Connecting;
@@ -14,6 +8,12 @@ using MQTTnet.Exceptions;
using MQTTnet.Protocol;
using MQTTnet.Server;
using MQTTnet.Tests.Mockups;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.Tests
{
@@ -29,9 +29,9 @@ namespace MQTTnet.Tests
{
await testEnvironment.StartServerAsync();
var client = await testEnvironment.ConnectClientAsync();
await client.SubscribeAsync("#");
var replyReceived = false;
client.UseApplicationMessageReceivedHandler(c =>
@@ -78,7 +78,7 @@ namespace MQTTnet.Tests
}
});
client2.UseApplicationMessageReceivedHandler(async c =>{ await client2.PublishAsync("reply", null, MqttQualityOfServiceLevel.AtLeastOnce); });
client2.UseApplicationMessageReceivedHandler(async c => { await client2.PublishAsync("reply", null, MqttQualityOfServiceLevel.AtLeastOnce); });
await client1.PublishAsync("request", null, MqttQualityOfServiceLevel.AtLeastOnce);
@@ -181,7 +181,7 @@ namespace MQTTnet.Tests
catch
{
}
SpinWait.SpinUntil(() => tries >= maxTries, 10000);
Assert.AreEqual(maxTries, tries);
@@ -215,7 +215,7 @@ namespace MQTTnet.Tests
Assert.AreEqual((ushort)4, result.PacketIdentifier);
}
}
[TestMethod]
public async Task Invalid_Connect_Throws_Exception()
{
@@ -558,6 +558,8 @@ namespace MQTTnet.Tests
clients.Add(await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithClientId("a")));
}
await Task.Delay(500);
var clientStatus = await testEnvironment.Server.GetClientStatusAsync();
var sessionStatus = await testEnvironment.Server.GetSessionStatusAsync();
@@ -565,7 +567,7 @@ namespace MQTTnet.Tests
{
Assert.IsFalse(clients[i].IsConnected);
}
Assert.IsTrue(clients[99].IsConnected);
Assert.AreEqual(1, clientStatus.Count);
@@ -583,7 +585,7 @@ namespace MQTTnet.Tests
var sendClient = await testEnvironment.ConnectClientAsync();
await sendClient.PublishAsync("x", "1");
await Task.Delay(100);
await Task.Delay(250);
Assert.AreEqual("1", receivedPayload);
}
Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs
+25 -23
View File
@@ -15,35 +15,36 @@ namespace MQTTnet.Tests
{
var factory = new MqttFactory();
//This test compares
//1. correct logID
// This test compares
// 1. correct logID
var logId = "logId";
string invalidLogId = null;
var hasInvalidLogId = false;
//2. if the total log calls are the same for global and local
var globalLogCount = 0;
// 2. if the total log calls are the same for global and local
//var globalLogCount = 0;
var localLogCount = 0;
var logger = new MqttNetLogger(logId);
//we have a theoretical bug here if a concurrent test is also logging
var globalLog = new EventHandler<MqttNetLogMessagePublishedEventArgs>((s, e) =>
{
if (logId != e.LogMessage.LogId)
{
invalidLogId = e.LogMessage.LogId;
}
// TODO: This is commented out because it is affected by other tests.
//// we have a theoretical bug here if a concurrent test is also logging
//var globalLog = new EventHandler<MqttNetLogMessagePublishedEventArgs>((s, e) =>
//{
// if (e.TraceMessage.LogId != logId)
// {
// invalidLogId = e.TraceMessage.LogId;
// }
Interlocked.Increment(ref globalLogCount);
});
// Interlocked.Increment(ref globalLogCount);
//});
MqttNetGlobalLogger.LogMessagePublished += globalLog;
//MqttNetGlobalLogger.LogMessagePublished += globalLog;
logger.LogMessagePublished += (s, e) =>
{
if (logId != e.LogMessage.LogId)
if (e.LogMessage.LogId != logId)
{
invalidLogId = e.LogMessage.LogId;
hasInvalidLogId = true;
}
Interlocked.Increment(ref localLogCount);
@@ -56,10 +57,10 @@ namespace MQTTnet.Tests
clientOptions.WithClientOptions(o => o.WithTcpServer("this_is_an_invalid_host").WithCommunicationTimeout(TimeSpan.FromSeconds(1)));
//try connect to get some log entries
// try connect to get some log entries
await managedClient.StartAsync(clientOptions.Build());
//wait at least connect timeout or we have some log messages
// wait at least connect timeout or we have some log messages
var tcs = new TaskCompletionSource<object>();
managedClient.ConnectingFailedHandler = new ConnectingFailedHandlerDelegate(e => tcs.TrySetResult(null));
await Task.WhenAny(Task.Delay(managedClient.Options.ClientOptions.CommunicationTimeout), tcs.Task);
@@ -68,12 +69,13 @@ namespace MQTTnet.Tests
{
await managedClient.StopAsync();
MqttNetGlobalLogger.LogMessagePublished -= globalLog;
//MqttNetGlobalLogger.LogMessagePublished -= globalLog;
}
Assert.IsNull(invalidLogId);
Assert.AreNotEqual(0, globalLogCount);
Assert.AreEqual(globalLogCount, localLogCount);
await Task.Delay(500);
Assert.IsFalse(hasInvalidLogId);
Assert.AreNotEqual(0, localLogCount);
}
}
}
Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitor_Tests.cs
+7 -7
View File
@@ -1,12 +1,12 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Server;
using MQTTnet.Server.Status;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.Tests
{
@@ -23,7 +23,7 @@ namespace MQTTnet.Tests
counter++;
return Task.CompletedTask;
},
new MqttNetLogger().CreateChildLogger());
new MqttNetLogger());
Assert.AreEqual(0, counter);
@@ -46,7 +46,7 @@ namespace MQTTnet.Tests
counter++;
return Task.CompletedTask;
},
new MqttNetLogger().CreateChildLogger());
new MqttNetLogger());
Assert.AreEqual(0, counter);
Tests/MQTTnet.Core.Tests/MqttNetLogger_Tests.cs
+50
View File
@@ -0,0 +1,50 @@
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Diagnostics;
namespace MQTTnet.Tests
{
[TestClass]
public class MqttNetLogger_Tests
{
[TestMethod]
public void Root_Log_Messages()
{
var logger = new MqttNetLogger();
var logMessagesCount = 0;
logger.LogMessagePublished += (s, e) =>
{
logMessagesCount++;
};
logger.Verbose("Verbose");
logger.Info("Info");
logger.Warning(null, "Warning");
logger.Error(null, "Error");
Assert.AreEqual(4, logMessagesCount);
}
[TestMethod]
public void Bubbling_Log_Messages()
{
var logger = new MqttNetLogger();
var childLogger = logger.CreateChildLogger("Source1");
var logMessagesCount = 0;
logger.LogMessagePublished += (s, e) =>
{
logMessagesCount++;
};
childLogger.Verbose("Verbose");
childLogger.Info("Info");
childLogger.Warning(null, "Warning");
childLogger.Error(null, "Error");
Assert.AreEqual(4, logMessagesCount);
}
}
}
Tests/MQTTnet.Core.Tests/Server_Tests.cs
+1 -1
View File
@@ -1050,6 +1050,7 @@ namespace MQTTnet.Tests
Assert.AreEqual("c", flow);
// dc
// Connect client with same client ID. Should disconnect existing client.
var c2 = await testEnvironment.ConnectClientAsync(clientOptionsBuilder);
c2.UseApplicationMessageReceivedHandler(_ =>
@@ -1075,7 +1076,6 @@ namespace MQTTnet.Tests
flow = string.Join(string.Empty, events);
Assert.AreEqual("cdcr", flow);
// nothing
Assert.AreEqual(false, c1.IsConnected);
Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs
+27 -17
View File
@@ -1,12 +1,4 @@
using System;
using System.Collections.Concurrent;
using System.Collections.ObjectModel;
using System.Text;
using System.Threading.Tasks;
using Windows.Security.Cryptography.Certificates;
using Windows.UI.Core;
using Windows.UI.Xaml;
using MQTTnet.Client;
using MQTTnet.Client;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
@@ -14,14 +6,22 @@ using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Extensions.Rpc;
using MQTTnet.Extensions.WebSocket4Net;
using MQTTnet.Formatter;
using MQTTnet.Implementations;
using MQTTnet.Protocol;
using MQTTnet.Server;
using MQTTnet.Server.Status;
using System;
using System.Collections.Concurrent;
using System.Collections.ObjectModel;
using System.Text;
using System.Threading.Tasks;
using Windows.Security.Cryptography.Certificates;
using Windows.UI.Core;
using Windows.UI.Xaml;
using MqttClientConnectedEventArgs = MQTTnet.Client.Connecting.MqttClientConnectedEventArgs;
using MqttClientDisconnectedEventArgs = MQTTnet.Client.Disconnecting.MqttClientDisconnectedEventArgs;
using MQTTnet.Extensions.WebSocket4Net;
namespace MQTTnet.TestApp.UniversalWindows
{
@@ -141,7 +141,7 @@ namespace MQTTnet.TestApp.UniversalWindows
Password = Encoding.UTF8.GetBytes(Password.Text)
};
}
options.CleanSession = CleanSession.IsChecked == true;
options.KeepAlivePeriod = TimeSpan.FromSeconds(double.Parse(KeepAliveInterval.Text));
@@ -198,16 +198,26 @@ namespace MQTTnet.TestApp.UniversalWindows
private void OnDisconnected(MqttClientDisconnectedEventArgs e)
{
_traceMessages.Enqueue(new MqttNetLogMessage("", DateTime.Now, -1,
"", MqttNetLogLevel.Info, "! DISCONNECTED EVENT FIRED", null));
_traceMessages.Enqueue(new MqttNetLogMessage
{
Timestamp = DateTime.UtcNow,
ThreadId = -1,
Level = MqttNetLogLevel.Info,
Message = "! DISCONNECTED EVENT FIRED",
});
Task.Run(UpdateLogAsync);
}
private void OnConnected(MqttClientConnectedEventArgs e)
{
_traceMessages.Enqueue(new MqttNetLogMessage("", DateTime.Now, -1,
"", MqttNetLogLevel.Info, "! CONNECTED EVENT FIRED", null));
_traceMessages.Enqueue(new MqttNetLogMessage
{
Timestamp = DateTime.UtcNow,
ThreadId = -1,
Level = MqttNetLogLevel.Info,
Message = "! CONNECTED EVENT FIRED",
});
Task.Run(UpdateLogAsync);
}
@@ -538,7 +548,7 @@ namespace MQTTnet.TestApp.UniversalWindows
{
//...
}
client.UseApplicationMessageReceivedHandler(e => Handler(e));
// Subscribe after connect
@@ -614,7 +624,7 @@ namespace MQTTnet.TestApp.UniversalWindows
};
}
}
// ----------------------------------
{
var options = new MqttServerOptions();