From 04ccdcd5628eb3f0e0338ceee73fc4b087745ec2 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 1 Apr 2020 20:43:00 +0200 Subject: [PATCH 1/5] Add low level client. --- Build/MQTTnet.nuspec | 1 + .../MqttClientConnectionContextFactory.cs | 9 +- .../MqttWebSocketServerAdapter.cs | 14 +- .../ManagedMqttClient.cs | 18 +-- .../WebSocket4NetMqttClientAdapterFactory.cs | 24 ++-- .../Logging/MqttNetChildLoggerWrapper.cs | 70 +++++----- .../Logging/MqttNetLoggerWrapper.cs | 19 ++- .../Adapter/IMqttClientAdapterFactory.cs | 2 +- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 66 ++++----- Source/MQTTnet/Client/MqttClient.cs | 10 +- .../Options/MqttClientOptionsBuilder.cs | 16 +-- .../Diagnostics/IMqttNetChildLogger.cs | 17 --- Source/MQTTnet/Diagnostics/IMqttNetLogger.cs | 4 +- .../MQTTnet/Diagnostics/MqttNetChildLogger.cs | 51 ------- .../MqttNetLogMessagePublishedEventArgs.cs | 4 + Source/MQTTnet/Diagnostics/MqttNetLogger.cs | 18 ++- .../Diagnostics/MqttNetLoggerExtensions.cs | 35 +++++ ...Provider.cs => TargetFrameworkProvider.cs} | 2 +- .../MqttClientAdapterFactory.cs | 8 +- .../MqttTcpServerAdapter.Uwp.cs | 12 +- .../Implementations/MqttTcpServerAdapter.cs | 43 +++--- .../Implementations/MqttTcpServerListener.cs | 44 +++--- Source/MQTTnet/Internal/TaskExtensions.cs | 6 +- .../LowLevelClient/ILowLevelMqttClient.cs | 19 +++ .../LowLevelClient/LowLevelMqttClient.cs | 128 ++++++++++++++++++ Source/MQTTnet/MqttFactory.cs | 42 +++++- .../Server/IMqttRetainedMessagesManager.cs | 6 +- Source/MQTTnet/Server/MqttClientConnection.cs | 14 +- .../Server/MqttClientKeepAliveMonitor.cs | 14 +- Source/MQTTnet/Server/MqttClientSession.cs | 10 +- .../Server/MqttClientSessionsManager.cs | 16 +-- .../Server/MqttRetainedMessagesManager.cs | 12 +- Source/MQTTnet/Server/MqttServer.cs | 4 +- .../Server/MqttServerEventDispatcher.cs | 10 +- Source/MQTTnet/TopicFilter.cs | 17 +-- Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs | 2 +- Tests/MQTTnet.Benchmarks/Program.cs | 2 +- .../LowLevelMqttClient_Tests.cs | 111 +++++++++++++++ .../Mockups/TestClientWrapper.cs | 49 ++++--- .../Mockups/TestEnvironment.cs | 51 ++++--- .../MQTTnet.Core.Tests/Mockups/TestLogger.cs | 20 +-- .../TestMqttCommunicationAdapterFactory.cs | 4 +- .../Mockups/TestServerWrapper.cs | 49 ++++--- Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs | 18 +-- Tests/MQTTnet.Core.Tests/Server_Tests.cs | 36 ++--- Tests/MQTTnet.TestApp.NetCore/Program.cs | 2 +- 46 files changed, 711 insertions(+), 418 deletions(-) delete mode 100644 Source/MQTTnet/Diagnostics/IMqttNetChildLogger.cs delete mode 100644 Source/MQTTnet/Diagnostics/MqttNetChildLogger.cs create mode 100644 Source/MQTTnet/Diagnostics/MqttNetLoggerExtensions.cs rename Source/MQTTnet/Diagnostics/{TargetFrameworkInfoProvider.cs => TargetFrameworkProvider.cs} (89%) create mode 100644 Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs create mode 100644 Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs create mode 100644 Tests/MQTTnet.Core.Tests/LowLevelMqttClient_Tests.cs diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index fd8614e5..318faddc 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -21,6 +21,7 @@ * [Server] Added interceptor for unsubscriptions. * [MQTTnet.Server] Added interceptor for unsubscriptions. * [MQTTnet.AspNetCore] improved compatibility with AspNetCore 3.1 +* [LowLevelMqttClient] Added low level MQTT client in order to provide more flexibility when using the MQTT protocol. This client requires detailed knowledge about the MQTT protocol. Copyright Christian Kratky 2016-2019 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin diff --git a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs index e8f570c2..44acb777 100644 --- a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs +++ b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs @@ -1,17 +1,16 @@ -using System; -using System.Net; -using MQTTnet.Adapter; +using MQTTnet.Adapter; using MQTTnet.AspNetCore.Client.Tcp; -using MQTTnet.Client; using MQTTnet.Client.Options; using MQTTnet.Diagnostics; using MQTTnet.Formatter; +using System; +using System.Net; namespace MQTTnet.AspNetCore.Client { public class MqttClientConnectionContextFactory : IMqttClientAdapterFactory { - public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger) + public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger) { if (options == null) throw new ArgumentNullException(nameof(options)); diff --git a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs index a052dae0..8e0134c8 100644 --- a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs +++ b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs @@ -1,20 +1,20 @@ -using System; -using System.Net.WebSockets; -using System.Threading.Tasks; -using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http; using MQTTnet.Adapter; using MQTTnet.Diagnostics; using MQTTnet.Formatter; using MQTTnet.Implementations; using MQTTnet.Server; +using System; +using System.Net.WebSockets; +using System.Threading.Tasks; namespace MQTTnet.AspNetCore { public class MqttWebSocketServerAdapter : IMqttServerAdapter { - private readonly IMqttNetChildLogger _logger; + private readonly IMqttNetLogger _logger; - public MqttWebSocketServerAdapter(IMqttNetChildLogger logger) + public MqttWebSocketServerAdapter(IMqttNetLogger logger) { if (logger == null) throw new ArgumentNullException(nameof(logger)); @@ -38,7 +38,7 @@ namespace MQTTnet.AspNetCore if (webSocket == null) throw new ArgumentNullException(nameof(webSocket)); var endpoint = $"{httpContext.Connection.RemoteIpAddress}:{httpContext.Connection.RemotePort}"; - + var clientCertificate = await httpContext.Connection.GetClientCertificateAsync().ConfigureAwait(false); try { diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 8b5a48f7..10d7d77c 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -1,9 +1,4 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; -using MQTTnet.Client; +using MQTTnet.Client; using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Publishing; @@ -13,6 +8,11 @@ using MQTTnet.Exceptions; using MQTTnet.Internal; using MQTTnet.Protocol; using MQTTnet.Server; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Extensions.ManagedClient { @@ -33,7 +33,7 @@ namespace MQTTnet.Extensions.ManagedClient private readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0); private readonly IMqttClient _mqttClient; - private readonly IMqttNetChildLogger _logger; + private readonly IMqttNetLogger _logger; private readonly AsyncLock _messageQueueLock = new AsyncLock(); @@ -42,8 +42,8 @@ namespace MQTTnet.Extensions.ManagedClient private Task _maintainConnectionTask; private ManagedMqttClientStorageManager _storageManager; - - public ManagedMqttClient(IMqttClient mqttClient, IMqttNetChildLogger logger) + + public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger) { _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient)); diff --git a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs index 7625fc92..1c3a9c68 100644 --- a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs +++ b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs @@ -1,15 +1,15 @@ -using System; -using MQTTnet.Adapter; +using MQTTnet.Adapter; using MQTTnet.Client.Options; using MQTTnet.Diagnostics; using MQTTnet.Formatter; using MQTTnet.Implementations; +using System; namespace MQTTnet.Extensions.WebSocket4Net { public class WebSocket4NetMqttClientAdapterFactory : IMqttClientAdapterFactory { - public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger) + public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger) { if (options == null) throw new ArgumentNullException(nameof(options)); if (logger == null) throw new ArgumentNullException(nameof(logger)); @@ -17,19 +17,19 @@ namespace MQTTnet.Extensions.WebSocket4Net switch (options.ChannelOptions) { case MqttClientTcpOptions _: - { - return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger); - } + { + return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger); + } case MqttClientWebSocketOptions webSocketOptions: - { - return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options, webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger); - } + { + return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options, webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger); + } default: - { - throw new NotSupportedException(); - } + { + throw new NotSupportedException(); + } } } } diff --git a/Source/MQTTnet.Server/Logging/MqttNetChildLoggerWrapper.cs b/Source/MQTTnet.Server/Logging/MqttNetChildLoggerWrapper.cs index 7f351178..656239fb 100644 --- a/Source/MQTTnet.Server/Logging/MqttNetChildLoggerWrapper.cs +++ b/Source/MQTTnet.Server/Logging/MqttNetChildLoggerWrapper.cs @@ -1,43 +1,43 @@ -using System; -using MQTTnet.Diagnostics; +//using MQTTnet.Diagnostics; +//using System; -namespace MQTTnet.Server.Logging -{ - public class MqttNetChildLoggerWrapper : IMqttNetChildLogger - { - private readonly MqttNetLoggerWrapper _logger; - private readonly string _source; +//namespace MQTTnet.Server.Logging +//{ +// public class MqttNetChildLoggerWrapper : IMqttNetChildLogger +// { +// private readonly MqttNetLoggerWrapper _logger; +// private readonly string _source; - public MqttNetChildLoggerWrapper(string source, MqttNetLoggerWrapper logger) - { - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); +// public MqttNetChildLoggerWrapper(string source, MqttNetLoggerWrapper logger) +// { +// _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - _source = source; - } +// _source = source; +// } - public IMqttNetChildLogger CreateChildLogger(string source = null) - { - return _logger.CreateChildLogger(source); - } +// public IMqttNetLogger CreateChildLogger(string source = null) +// { +// return _logger.CreateChildLogger(source); +// } - public void Verbose(string message, params object[] parameters) - { - _logger.Publish(MqttNetLogLevel.Verbose, _source, message, parameters, null); - } +// public void Verbose(string message, params object[] parameters) +// { +// _logger.Publish(MqttNetLogLevel.Verbose, _source, message, parameters, null); +// } - public void Info(string message, params object[] parameters) - { - _logger.Publish(MqttNetLogLevel.Info, _source, message, parameters, null); - } +// public void Info(string message, params object[] parameters) +// { +// _logger.Publish(MqttNetLogLevel.Info, _source, message, parameters, null); +// } - public void Warning(Exception exception, string message, params object[] parameters) - { - _logger.Publish(MqttNetLogLevel.Warning, _source, message, parameters, exception); - } +// public void Warning(Exception exception, string message, params object[] parameters) +// { +// _logger.Publish(MqttNetLogLevel.Warning, _source, message, parameters, exception); +// } - public void Error(Exception exception, string message, params object[] parameters) - { - _logger.Publish(MqttNetLogLevel.Error, _source, message, parameters, exception); - } - } -} +// public void Error(Exception exception, string message, params object[] parameters) +// { +// _logger.Publish(MqttNetLogLevel.Error, _source, message, parameters, exception); +// } +// } +//} diff --git a/Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs b/Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs index 3f521092..668ae265 100644 --- a/Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs +++ b/Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs @@ -1,7 +1,7 @@ -using System; -using System.Threading; -using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging; using MQTTnet.Diagnostics; +using System; +using System.Threading; namespace MQTTnet.Server.Logging { @@ -16,9 +16,9 @@ namespace MQTTnet.Server.Logging public event EventHandler LogMessagePublished; - public IMqttNetChildLogger CreateChildLogger(string source = null) + public IMqttNetLogger CreateChildLogger(string source = null) { - return new MqttNetChildLoggerWrapper(source, this); + return new MqttNetLogger(source); } public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception) @@ -33,8 +33,13 @@ namespace MQTTnet.Server.Logging logMessagePublishedEvent.Invoke(this, new MqttNetLogMessagePublishedEventArgs(logMessage)); } } - - private static LogLevel ConvertLogLevel(MqttNetLogLevel logLevel) + + public void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception) + { + Publish(logLevel, null, message, parameters, exception); + } + + static LogLevel ConvertLogLevel(MqttNetLogLevel logLevel) { switch (logLevel) { diff --git a/Source/MQTTnet/Adapter/IMqttClientAdapterFactory.cs b/Source/MQTTnet/Adapter/IMqttClientAdapterFactory.cs index cb0988d8..c9aba6f2 100644 --- a/Source/MQTTnet/Adapter/IMqttClientAdapterFactory.cs +++ b/Source/MQTTnet/Adapter/IMqttClientAdapterFactory.cs @@ -5,6 +5,6 @@ namespace MQTTnet.Adapter { public interface IMqttClientAdapterFactory { - IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger); + IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger); } } diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index d7585cb2..5dd02815 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -1,3 +1,9 @@ +using MQTTnet.Channel; +using MQTTnet.Diagnostics; +using MQTTnet.Exceptions; +using MQTTnet.Formatter; +using MQTTnet.Internal; +using MQTTnet.Packets; using System; using System.IO; using System.Net.Sockets; @@ -5,32 +11,26 @@ using System.Runtime.InteropServices; using System.Security.Cryptography.X509Certificates; using System.Threading; using System.Threading.Tasks; -using MQTTnet.Channel; -using MQTTnet.Diagnostics; -using MQTTnet.Exceptions; -using MQTTnet.Formatter; -using MQTTnet.Internal; -using MQTTnet.Packets; namespace MQTTnet.Adapter { - public class MqttChannelAdapter : Disposable, IMqttChannelAdapter + public sealed class MqttChannelAdapter : Disposable, IMqttChannelAdapter { - private const uint ErrorOperationAborted = 0x800703E3; - private const int ReadBufferSize = 4096; // TODO: Move buffer size to config + const uint ErrorOperationAborted = 0x800703E3; + const int ReadBufferSize = 4096; // TODO: Move buffer size to config - private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1); + readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1); - private readonly IMqttNetChildLogger _logger; - private readonly IMqttChannel _channel; - private readonly MqttPacketReader _packetReader; + readonly IMqttNetLogger _logger; + readonly IMqttChannel _channel; + readonly MqttPacketReader _packetReader; - private readonly byte[] _fixedHeaderBuffer = new byte[2]; - - private long _bytesReceived; - private long _bytesSent; + readonly byte[] _fixedHeaderBuffer = new byte[2]; - public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, IMqttNetChildLogger logger) + long _bytesReceived; + long _bytesSent; + + public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, IMqttNetLogger logger) { if (logger == null) throw new ArgumentNullException(nameof(logger)); @@ -55,7 +55,7 @@ namespace MQTTnet.Adapter public Action ReadingPacketStartedCallback { get; set; } public Action ReadingPacketCompletedCallback { get; set; } - + public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) { ThrowIfDisposed(); @@ -207,7 +207,18 @@ namespace MQTTnet.Adapter Interlocked.Exchange(ref _bytesSent, 0L); } - private async Task ReceiveAsync(CancellationToken cancellationToken) + protected override void Dispose(bool disposing) + { + if (disposing) + { + _channel?.Dispose(); + _writerSemaphore?.Dispose(); + } + + base.Dispose(disposing); + } + + async Task ReceiveAsync(CancellationToken cancellationToken) { var readFixedHeaderResult = await _packetReader.ReadFixedHeaderAsync(_fixedHeaderBuffer, cancellationToken).ConfigureAwait(false); @@ -267,25 +278,14 @@ namespace MQTTnet.Adapter } } - protected override void Dispose(bool disposing) - { - if (disposing) - { - _channel?.Dispose(); - _writerSemaphore?.Dispose(); - } - - base.Dispose(disposing); - } - - private static bool IsWrappedException(Exception exception) + static bool IsWrappedException(Exception exception) { return exception is OperationCanceledException || exception is MqttCommunicationTimedOutException || exception is MqttCommunicationException; } - private static void WrapException(Exception exception) + static void WrapException(Exception exception) { if (exception is IOException && exception.InnerException is SocketException innerException) { diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 29687fc0..2546fc1d 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -1,7 +1,3 @@ -using System; -using System.Diagnostics; -using System.Threading; -using System.Threading.Tasks; using MQTTnet.Adapter; using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; @@ -17,6 +13,10 @@ using MQTTnet.Internal; using MQTTnet.PacketDispatcher; using MQTTnet.Packets; using MQTTnet.Protocol; +using System; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Client { @@ -29,7 +29,7 @@ namespace MQTTnet.Client private readonly object _disconnectLock = new object(); private readonly IMqttClientAdapterFactory _adapterFactory; - private readonly IMqttNetChildLogger _logger; + private readonly IMqttNetLogger _logger; private CancellationTokenSource _backgroundCancellationTokenSource; private Task _packetReceiverTask; diff --git a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs index 4fd0ccfc..b3018190 100644 --- a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs @@ -1,19 +1,19 @@ -using System; +using MQTTnet.Client.ExtendedAuthenticationExchange; +using MQTTnet.Formatter; +using System; using System.Linq; using System.Text; -using MQTTnet.Client.ExtendedAuthenticationExchange; -using MQTTnet.Formatter; namespace MQTTnet.Client.Options { public class MqttClientOptionsBuilder { - private readonly MqttClientOptions _options = new MqttClientOptions(); + readonly MqttClientOptions _options = new MqttClientOptions(); - private MqttClientTcpOptions _tcpOptions; - private MqttClientWebSocketOptions _webSocketOptions; - private MqttClientOptionsBuilderTlsParameters _tlsParameters; - private MqttClientWebSocketProxyOptions _proxyOptions; + MqttClientTcpOptions _tcpOptions; + MqttClientWebSocketOptions _webSocketOptions; + MqttClientOptionsBuilderTlsParameters _tlsParameters; + MqttClientWebSocketProxyOptions _proxyOptions; public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value) { diff --git a/Source/MQTTnet/Diagnostics/IMqttNetChildLogger.cs b/Source/MQTTnet/Diagnostics/IMqttNetChildLogger.cs deleted file mode 100644 index df99f443..00000000 --- a/Source/MQTTnet/Diagnostics/IMqttNetChildLogger.cs +++ /dev/null @@ -1,17 +0,0 @@ -using System; - -namespace MQTTnet.Diagnostics -{ - public interface IMqttNetChildLogger - { - IMqttNetChildLogger CreateChildLogger(string source = null); - - void Verbose(string message, params object[] parameters); - - void Info(string message, params object[] parameters); - - void Warning(Exception exception, string message, params object[] parameters); - - void Error(Exception exception, string message, params object[] parameters); - } -} diff --git a/Source/MQTTnet/Diagnostics/IMqttNetLogger.cs b/Source/MQTTnet/Diagnostics/IMqttNetLogger.cs index 73b6fcbd..40bb04b5 100644 --- a/Source/MQTTnet/Diagnostics/IMqttNetLogger.cs +++ b/Source/MQTTnet/Diagnostics/IMqttNetLogger.cs @@ -6,8 +6,8 @@ namespace MQTTnet.Diagnostics { event EventHandler LogMessagePublished; - IMqttNetChildLogger CreateChildLogger(string source = null); + IMqttNetLogger CreateChildLogger(string source = null); - void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception); + void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception); } } diff --git a/Source/MQTTnet/Diagnostics/MqttNetChildLogger.cs b/Source/MQTTnet/Diagnostics/MqttNetChildLogger.cs deleted file mode 100644 index 37334543..00000000 --- a/Source/MQTTnet/Diagnostics/MqttNetChildLogger.cs +++ /dev/null @@ -1,51 +0,0 @@ -using System; - -namespace MQTTnet.Diagnostics -{ - public class MqttNetChildLogger : IMqttNetChildLogger - { - private readonly IMqttNetLogger _logger; - private readonly string _source; - - public MqttNetChildLogger(IMqttNetLogger logger, string source) - { - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - _source = source; - } - - public IMqttNetChildLogger CreateChildLogger(string source) - { - string childSource; - if (!string.IsNullOrEmpty(_source)) - { - childSource = _source + "." + source; - } - else - { - childSource = source; - } - - return new MqttNetChildLogger(_logger, childSource); - } - - public void Verbose(string message, params object[] parameters) - { - _logger.Publish(MqttNetLogLevel.Verbose, _source, message, parameters, null); - } - - public void Info(string message, params object[] parameters) - { - _logger.Publish(MqttNetLogLevel.Info, _source, message, parameters, null); - } - - public void Warning(Exception exception, string message, params object[] parameters) - { - _logger.Publish(MqttNetLogLevel.Warning, _source, message, parameters, exception); - } - - public void Error(Exception exception, string message, params object[] parameters) - { - _logger.Publish(MqttNetLogLevel.Error, _source, message, parameters, exception); - } - } -} diff --git a/Source/MQTTnet/Diagnostics/MqttNetLogMessagePublishedEventArgs.cs b/Source/MQTTnet/Diagnostics/MqttNetLogMessagePublishedEventArgs.cs index 8a0c0fcb..1b3488e1 100644 --- a/Source/MQTTnet/Diagnostics/MqttNetLogMessagePublishedEventArgs.cs +++ b/Source/MQTTnet/Diagnostics/MqttNetLogMessagePublishedEventArgs.cs @@ -7,8 +7,12 @@ namespace MQTTnet.Diagnostics public MqttNetLogMessagePublishedEventArgs(MqttNetLogMessage logMessage) { TraceMessage = logMessage ?? throw new ArgumentNullException(nameof(logMessage)); + LogMessage = logMessage ?? throw new ArgumentNullException(nameof(logMessage)); } + [Obsolete("Use new proeprty LogMessage instead.")] public MqttNetLogMessage TraceMessage { get; } + + public MqttNetLogMessage LogMessage { get; } } } diff --git a/Source/MQTTnet/Diagnostics/MqttNetLogger.cs b/Source/MQTTnet/Diagnostics/MqttNetLogger.cs index 56a664e4..b14bf343 100644 --- a/Source/MQTTnet/Diagnostics/MqttNetLogger.cs +++ b/Source/MQTTnet/Diagnostics/MqttNetLogger.cs @@ -4,21 +4,27 @@ namespace MQTTnet.Diagnostics { public class MqttNetLogger : IMqttNetLogger { - private readonly string _logId; + readonly string _logId; + readonly string _source; - public MqttNetLogger(string logId = null) + public MqttNetLogger(string source, string logId = null) { + _source = source; _logId = logId; } + public MqttNetLogger() + { + } + public event EventHandler LogMessagePublished; - public IMqttNetChildLogger CreateChildLogger(string source = null) + public IMqttNetLogger CreateChildLogger(string source = null) { - return new MqttNetChildLogger(this, source); + return new MqttNetLogger(source, _logId); } - public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception) + public void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception) { var hasLocalListeners = LogMessagePublished != null; var hasGlobalListeners = MqttNetGlobalLogger.HasListeners; @@ -40,7 +46,7 @@ namespace MQTTnet.Diagnostics } } - var traceMessage = new MqttNetLogMessage(_logId, DateTime.UtcNow, Environment.CurrentManagedThreadId, source, logLevel, message, exception); + var traceMessage = new MqttNetLogMessage(_logId, DateTime.UtcNow, Environment.CurrentManagedThreadId, _source, logLevel, message, exception); if (hasGlobalListeners) { diff --git a/Source/MQTTnet/Diagnostics/MqttNetLoggerExtensions.cs b/Source/MQTTnet/Diagnostics/MqttNetLoggerExtensions.cs new file mode 100644 index 00000000..bc8a97b6 --- /dev/null +++ b/Source/MQTTnet/Diagnostics/MqttNetLoggerExtensions.cs @@ -0,0 +1,35 @@ +using System; + +namespace MQTTnet.Diagnostics +{ + public static class MqttNetLoggerExtensions + { + public static void Verbose(this IMqttNetLogger logger, string message, params object[] parameters) + { + if (logger is null) throw new ArgumentNullException(nameof(logger)); + + logger.Publish(MqttNetLogLevel.Verbose, message, parameters, null); + } + + public static void Info(this IMqttNetLogger logger, string message, params object[] parameters) + { + if (logger is null) throw new ArgumentNullException(nameof(logger)); + + logger.Publish(MqttNetLogLevel.Info, message, parameters, null); + } + + public static void Warning(this IMqttNetLogger logger, Exception exception, string message, params object[] parameters) + { + if (logger is null) throw new ArgumentNullException(nameof(logger)); + + logger.Publish(MqttNetLogLevel.Warning, message, parameters, exception); + } + + public static void Error(this IMqttNetLogger logger, Exception exception, string message, params object[] parameters) + { + if (logger is null) throw new ArgumentNullException(nameof(logger)); + + logger.Publish(MqttNetLogLevel.Error, message, parameters, exception); + } + } +} diff --git a/Source/MQTTnet/Diagnostics/TargetFrameworkInfoProvider.cs b/Source/MQTTnet/Diagnostics/TargetFrameworkProvider.cs similarity index 89% rename from Source/MQTTnet/Diagnostics/TargetFrameworkInfoProvider.cs rename to Source/MQTTnet/Diagnostics/TargetFrameworkProvider.cs index efbf08b8..7135af67 100644 --- a/Source/MQTTnet/Diagnostics/TargetFrameworkInfoProvider.cs +++ b/Source/MQTTnet/Diagnostics/TargetFrameworkProvider.cs @@ -1,6 +1,6 @@ namespace MQTTnet.Diagnostics { - public static class TargetFrameworkInfoProvider + public static class TargetFrameworkProvider { public static string TargetFramework { diff --git a/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs b/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs index 2377cde7..148fe044 100644 --- a/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs +++ b/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs @@ -1,17 +1,17 @@ -using System; -using MQTTnet.Adapter; +using MQTTnet.Adapter; using MQTTnet.Client.Options; using MQTTnet.Diagnostics; using MQTTnet.Formatter; +using System; namespace MQTTnet.Implementations { public class MqttClientAdapterFactory : IMqttClientAdapterFactory { - public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger) + public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger) { if (options == null) throw new ArgumentNullException(nameof(options)); - + switch (options.ChannelOptions) { case MqttClientTcpOptions _: diff --git a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs index 3b24bd13..4d26da7c 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs @@ -11,14 +11,14 @@ using System.Threading.Tasks; namespace MQTTnet.Implementations { - public class MqttTcpServerAdapter : IMqttServerAdapter + public sealed class MqttTcpServerAdapter : IMqttServerAdapter { - private readonly IMqttNetChildLogger _logger; + readonly IMqttNetLogger _logger; - private IMqttServerOptions _options; - private StreamSocketListener _listener; + IMqttServerOptions _options; + StreamSocketListener _listener; - public MqttTcpServerAdapter(IMqttNetChildLogger logger) + public MqttTcpServerAdapter(IMqttNetLogger logger) { if (logger == null) throw new ArgumentNullException(nameof(logger)); @@ -68,7 +68,7 @@ namespace MQTTnet.Implementations _listener = null; } - private async void OnConnectionReceivedAsync(StreamSocketListener sender, StreamSocketListenerConnectionReceivedEventArgs args) + async void OnConnectionReceivedAsync(StreamSocketListener sender, StreamSocketListenerConnectionReceivedEventArgs args) { try { diff --git a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs index 501c4da6..22a5824c 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs @@ -1,4 +1,8 @@ #if !WINDOWS_UWP +using MQTTnet.Adapter; +using MQTTnet.Diagnostics; +using MQTTnet.Internal; +using MQTTnet.Server; using System; using System.Collections.Generic; using System.Net; @@ -6,21 +10,17 @@ using System.Net.Sockets; using System.Security.Cryptography.X509Certificates; using System.Threading; using System.Threading.Tasks; -using MQTTnet.Adapter; -using MQTTnet.Diagnostics; -using MQTTnet.Internal; -using MQTTnet.Server; namespace MQTTnet.Implementations { - public class MqttTcpServerAdapter : Disposable, IMqttServerAdapter + public sealed class MqttTcpServerAdapter : Disposable, IMqttServerAdapter { - private readonly List _listeners = new List(); - private readonly IMqttNetChildLogger _logger; + readonly List _listeners = new List(); + readonly IMqttNetLogger _logger; - private CancellationTokenSource _cancellationTokenSource; + CancellationTokenSource _cancellationTokenSource; - public MqttTcpServerAdapter(IMqttNetChildLogger logger) + public MqttTcpServerAdapter(IMqttNetLogger logger) { if (logger == null) throw new ArgumentNullException(nameof(logger)); @@ -59,7 +59,7 @@ namespace MQTTnet.Implementations { tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate, options.TlsEndpointOptions.CertificateCredentials.Password); } - + if (!tlsCertificate.HasPrivateKey) { throw new InvalidOperationException("The certificate for TLS encryption must contain the private key."); @@ -77,7 +77,17 @@ namespace MQTTnet.Implementations return Task.FromResult(0); } - private void Cleanup() + protected override void Dispose(bool disposing) + { + if (disposing) + { + Cleanup(); + } + + base.Dispose(disposing); + } + + void Cleanup() { _cancellationTokenSource?.Cancel(false); _cancellationTokenSource?.Dispose(); @@ -91,16 +101,7 @@ namespace MQTTnet.Implementations _listeners.Clear(); } - protected override void Dispose(bool disposing) - { - if (disposing) - { - Cleanup(); - } - base.Dispose(disposing); - } - - private void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate, CancellationToken cancellationToken) + void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate, CancellationToken cancellationToken) { if (!options.BoundInterNetworkAddress.Equals(IPAddress.None)) { diff --git a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs index f2f439e8..84cd7b96 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs @@ -1,4 +1,9 @@ #if !WINDOWS_UWP +using MQTTnet.Adapter; +using MQTTnet.Diagnostics; +using MQTTnet.Formatter; +using MQTTnet.Internal; +using MQTTnet.Server; using System; using System.IO; using System.Net; @@ -7,21 +12,16 @@ using System.Net.Sockets; using System.Security.Cryptography.X509Certificates; using System.Threading; using System.Threading.Tasks; -using MQTTnet.Adapter; -using MQTTnet.Diagnostics; -using MQTTnet.Formatter; -using MQTTnet.Internal; -using MQTTnet.Server; namespace MQTTnet.Implementations { - public class MqttTcpServerListener : IDisposable + public sealed class MqttTcpServerListener : IDisposable { - private readonly IMqttNetChildLogger _logger; - private readonly AddressFamily _addressFamily; - private readonly MqttServerTcpEndpointBaseOptions _options; - private readonly MqttServerTlsTcpEndpointOptions _tlsOptions; - private readonly X509Certificate2 _tlsCertificate; + readonly IMqttNetLogger _logger; + readonly AddressFamily _addressFamily; + readonly MqttServerTcpEndpointBaseOptions _options; + readonly MqttServerTlsTcpEndpointOptions _tlsOptions; + readonly X509Certificate2 _tlsCertificate; private Socket _socket; private IPEndPoint _localEndPoint; @@ -30,7 +30,7 @@ namespace MQTTnet.Implementations AddressFamily addressFamily, MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate, - IMqttNetChildLogger logger) + IMqttNetLogger logger) { _addressFamily = addressFamily; _options = options; @@ -67,12 +67,12 @@ namespace MQTTnet.Implementations { _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true); } - + if (_options.NoDelay) { _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, true); } - + _socket.Bind(_localEndPoint); _socket.Listen(_options.ConnectionBacklog); @@ -87,7 +87,7 @@ namespace MQTTnet.Implementations throw; } - _logger.Warning(exception,"Error while creating listener socket for local end point '{0}'.", _localEndPoint); + _logger.Warning(exception, "Error while creating listener socket for local end point '{0}'.", _localEndPoint); return false; } } @@ -101,7 +101,7 @@ namespace MQTTnet.Implementations #endif } - private async Task AcceptClientConnectionsAsync(CancellationToken cancellationToken) + async Task AcceptClientConnectionsAsync(CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) { @@ -116,7 +116,7 @@ namespace MQTTnet.Implementations Task.Run(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken).Forget(_logger); } catch (OperationCanceledException) - { + { } catch (Exception exception) { @@ -128,14 +128,14 @@ namespace MQTTnet.Implementations continue; } } - + _logger.Error(exception, $"Error while accepting connection at TCP listener {_localEndPoint} TLS={_tlsCertificate != null}."); await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false); } } } - private async Task TryHandleClientConnectionAsync(Socket clientSocket) + async Task TryHandleClientConnectionAsync(Socket clientSocket) { Stream stream = null; string remoteEndPoint = null; @@ -160,9 +160,9 @@ namespace MQTTnet.Implementations var sslStream = new SslStream(stream, false, _tlsOptions.RemoteCertificateValidationCallback); await sslStream.AuthenticateAsServerAsync( - _tlsCertificate, - _tlsOptions.ClientCertificateRequired, - _tlsOptions.SslProtocol, + _tlsCertificate, + _tlsOptions.ClientCertificateRequired, + _tlsOptions.SslProtocol, _tlsOptions.CheckCertificateRevocation).ConfigureAwait(false); stream = sslStream; diff --git a/Source/MQTTnet/Internal/TaskExtensions.cs b/Source/MQTTnet/Internal/TaskExtensions.cs index 733631c5..12a8c072 100644 --- a/Source/MQTTnet/Internal/TaskExtensions.cs +++ b/Source/MQTTnet/Internal/TaskExtensions.cs @@ -1,11 +1,11 @@ -using System.Threading.Tasks; -using MQTTnet.Diagnostics; +using MQTTnet.Diagnostics; +using System.Threading.Tasks; namespace MQTTnet.Internal { public static class TaskExtensions { - public static void Forget(this Task task, IMqttNetChildLogger logger) + public static void Forget(this Task task, IMqttNetLogger logger) { task?.ContinueWith(t => { diff --git a/Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs b/Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs new file mode 100644 index 00000000..1734bf43 --- /dev/null +++ b/Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs @@ -0,0 +1,19 @@ +using MQTTnet.Client.Options; +using MQTTnet.Packets; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace MQTTnet.LowLevelClient +{ + public interface ILowLevelMqttClient : IDisposable + { + Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken); + + Task DisconnectAsync(CancellationToken cancellationToken); + + Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken); + + Task ReceiveAsync(CancellationToken cancellationToken); + } +} diff --git a/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs b/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs new file mode 100644 index 00000000..045100df --- /dev/null +++ b/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs @@ -0,0 +1,128 @@ +using MQTTnet.Adapter; +using MQTTnet.Client.Options; +using MQTTnet.Diagnostics; +using MQTTnet.Packets; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace MQTTnet.LowLevelClient +{ + public sealed class LowLevelMqttClient : ILowLevelMqttClient + { + readonly IMqttNetLogger _logger; + readonly IMqttClientAdapterFactory _clientAdapterFactory; + + IMqttChannelAdapter _adapter; + IMqttClientOptions _options; + + public LowLevelMqttClient(IMqttClientAdapterFactory clientAdapterFactory, IMqttNetLogger logger) + { + if (clientAdapterFactory is null) throw new ArgumentNullException(nameof(clientAdapterFactory)); + if (logger is null) throw new ArgumentNullException(nameof(logger)); + + _clientAdapterFactory = clientAdapterFactory; + _logger = logger.CreateChildLogger(nameof(LowLevelMqttClient)); + } + + bool IsConnected => _adapter != null; + + public async Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken) + { + if (options is null) throw new ArgumentNullException(nameof(options)); + + if (_adapter != null) + { + throw new InvalidOperationException("Low level MQTT client is already connected. Disconnect first before connecting again."); + } + + var newAdapter = _clientAdapterFactory.CreateClientAdapter(options, _logger); + + try + { + _logger.Verbose($"Trying to connect with server '{options.ChannelOptions}' (Timeout={options.CommunicationTimeout})."); + await newAdapter.ConnectAsync(options.CommunicationTimeout, cancellationToken).ConfigureAwait(false); + _logger.Verbose("Connection with server established."); + + _options = options; + } + catch (Exception) + { + _adapter.Dispose(); + throw; + } + + _adapter = newAdapter; + } + + public async Task DisconnectAsync(CancellationToken cancellationToken) + { + if (_adapter == null) + { + return; + } + + await SafeDisconnect(cancellationToken).ConfigureAwait(false); + _adapter = null; + } + + public async Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken) + { + if (packet is null) throw new ArgumentNullException(nameof(packet)); + + if (_adapter == null) + { + throw new InvalidOperationException("Low level MQTT client is not connected."); + } + + try + { + await _adapter.SendPacketAsync(packet, _options.CommunicationTimeout, cancellationToken).ConfigureAwait(false); + } + catch (Exception) + { + await SafeDisconnect(cancellationToken).ConfigureAwait(false); + throw; + } + } + + public async Task ReceiveAsync(CancellationToken cancellationToken) + { + if (_adapter == null) + { + throw new InvalidOperationException("Low level MQTT client is not connected."); + } + + try + { + return await _adapter.ReceivePacketAsync(_options.CommunicationTimeout, cancellationToken).ConfigureAwait(false); + } + catch (Exception) + { + await SafeDisconnect(cancellationToken).ConfigureAwait(false); + throw; + } + } + + public void Dispose() + { + _adapter?.Dispose(); + } + + async Task SafeDisconnect(CancellationToken cancellationToken) + { + try + { + await _adapter.DisconnectAsync(_options.CommunicationTimeout, cancellationToken).ConfigureAwait(false); + } + catch (Exception exception) + { + _logger.Error(exception, "Error while disconnecting."); + } + finally + { + _adapter.Dispose(); + } + } + } +} diff --git a/Source/MQTTnet/MqttFactory.cs b/Source/MQTTnet/MqttFactory.cs index cb835f60..da85cba7 100644 --- a/Source/MQTTnet/MqttFactory.cs +++ b/Source/MQTTnet/MqttFactory.cs @@ -1,18 +1,19 @@ -using System; -using System.Collections.Generic; -using MQTTnet.Adapter; +using MQTTnet.Adapter; using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Implementations; +using MQTTnet.LowLevelClient; using MQTTnet.Server; +using System; +using System.Collections.Generic; namespace MQTTnet { - public class MqttFactory : IMqttFactory + public sealed class MqttFactory : IMqttFactory { - private IMqttClientAdapterFactory _clientAdapterFactory = new MqttClientAdapterFactory(); + IMqttClientAdapterFactory _clientAdapterFactory = new MqttClientAdapterFactory(); - public MqttFactory() : this(new MqttNetLogger()) + public MqttFactory() : this(new MqttNetLogger(null, null)) { } @@ -29,6 +30,33 @@ namespace MQTTnet return this; } + public ILowLevelMqttClient CreateLowLevelMqttClient() + { + return CreateLowLevelMqttClient(DefaultLogger); + } + + public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger) + { + if (logger == null) throw new ArgumentNullException(nameof(logger)); + + return new LowLevelMqttClient(_clientAdapterFactory, logger); + } + + public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttClientAdapterFactory clientAdapterFactory) + { + if (clientAdapterFactory == null) throw new ArgumentNullException(nameof(clientAdapterFactory)); + + return new LowLevelMqttClient(_clientAdapterFactory, DefaultLogger); + } + + public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactoryy) + { + if (logger == null) throw new ArgumentNullException(nameof(logger)); + if (clientAdapterFactoryy == null) throw new ArgumentNullException(nameof(clientAdapterFactoryy)); + + return new LowLevelMqttClient(_clientAdapterFactory, logger); + } + public IMqttClient CreateMqttClient() { return CreateMqttClient(DefaultLogger); @@ -79,7 +107,7 @@ namespace MQTTnet public IMqttServer CreateMqttServer(IEnumerable serverAdapters) { if (serverAdapters == null) throw new ArgumentNullException(nameof(serverAdapters)); - + return new MqttServer(serverAdapters, DefaultLogger.CreateChildLogger()); } } diff --git a/Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs index 6ffdd2bf..ab1d38ce 100644 --- a/Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs @@ -1,12 +1,12 @@ -using System.Collections.Generic; +using MQTTnet.Diagnostics; +using System.Collections.Generic; using System.Threading.Tasks; -using MQTTnet.Diagnostics; namespace MQTTnet.Server { public interface IMqttRetainedMessagesManager { - Task Start(IMqttServerOptions options, IMqttNetChildLogger logger); + Task Start(IMqttServerOptions options, IMqttNetLogger logger); Task LoadMessagesAsync(); diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index c9e25535..a21873bd 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -1,8 +1,4 @@ -using System; -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; -using MQTTnet.Adapter; +using MQTTnet.Adapter; using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; @@ -12,6 +8,10 @@ using MQTTnet.PacketDispatcher; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server.Status; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Server { @@ -25,7 +25,7 @@ namespace MQTTnet.Server private readonly MqttClientKeepAliveMonitor _keepAliveMonitor; private readonly MqttClientSessionsManager _sessionsManager; - private readonly IMqttNetChildLogger _logger; + private readonly IMqttNetLogger _logger; private readonly IMqttServerOptions _serverOptions; private readonly IMqttChannelAdapter _channelAdapter; @@ -49,7 +49,7 @@ namespace MQTTnet.Server IMqttServerOptions serverOptions, MqttClientSessionsManager sessionsManager, IMqttRetainedMessagesManager retainedMessagesManager, - IMqttNetChildLogger logger) + IMqttNetLogger logger) { Session = session ?? throw new ArgumentNullException(nameof(session)); _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); diff --git a/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs b/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs index deefe4ce..9d17552c 100644 --- a/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs @@ -1,9 +1,9 @@ -using System; +using MQTTnet.Diagnostics; +using MQTTnet.Internal; +using System; using System.Diagnostics; using System.Threading; using System.Threading.Tasks; -using MQTTnet.Diagnostics; -using MQTTnet.Internal; namespace MQTTnet.Server { @@ -13,15 +13,15 @@ namespace MQTTnet.Server private readonly string _clientId; private readonly Func _keepAliveElapsedCallback; - private readonly IMqttNetChildLogger _logger; + private readonly IMqttNetLogger _logger; private bool _isPaused; - public MqttClientKeepAliveMonitor(string clientId, Func keepAliveElapsedCallback, IMqttNetChildLogger logger) + public MqttClientKeepAliveMonitor(string clientId, Func keepAliveElapsedCallback, IMqttNetLogger logger) { _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); _keepAliveElapsedCallback = keepAliveElapsedCallback ?? throw new ArgumentNullException(nameof(keepAliveElapsedCallback)); - + if (logger == null) throw new ArgumentNullException(nameof(logger)); _logger = logger.CreateChildLogger(nameof(MqttClientKeepAliveMonitor)); } @@ -32,7 +32,7 @@ namespace MQTTnet.Server { return; } - + Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken).Forget(_logger); } diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index d097b9f1..e0d21753 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -1,18 +1,18 @@ -using System; +using MQTTnet.Diagnostics; +using MQTTnet.Server.Status; +using System; using System.Collections.Generic; using System.Threading.Tasks; -using MQTTnet.Diagnostics; -using MQTTnet.Server.Status; namespace MQTTnet.Server { public class MqttClientSession { - private readonly IMqttNetChildLogger _logger; + private readonly IMqttNetLogger _logger; private readonly DateTime _createdTimestamp = DateTime.UtcNow; - public MqttClientSession(string clientId, IDictionary items, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetChildLogger logger) + public MqttClientSession(string clientId, IDictionary items, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetLogger logger) { ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); Items = items ?? throw new ArgumentNullException(nameof(items)); diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 28c163d9..dcbf5a4f 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -1,15 +1,15 @@ -using System; -using System.Collections.Concurrent; -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; -using MQTTnet.Adapter; +using MQTTnet.Adapter; using MQTTnet.Diagnostics; using MQTTnet.Formatter; using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server.Status; +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Server { @@ -27,14 +27,14 @@ namespace MQTTnet.Server private readonly IMqttRetainedMessagesManager _retainedMessagesManager; private readonly IMqttServerOptions _options; - private readonly IMqttNetChildLogger _logger; + private readonly IMqttNetLogger _logger; public MqttClientSessionsManager( IMqttServerOptions options, IMqttRetainedMessagesManager retainedMessagesManager, CancellationToken cancellationToken, MqttServerEventDispatcher eventDispatcher, - IMqttNetChildLogger logger) + IMqttNetLogger logger) { _cancellationToken = cancellationToken; diff --git a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs index f4ebe487..f6a994b9 100644 --- a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs @@ -1,10 +1,10 @@ -using System; +using MQTTnet.Diagnostics; +using MQTTnet.Implementations; +using MQTTnet.Internal; +using System; using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; -using MQTTnet.Diagnostics; -using MQTTnet.Implementations; -using MQTTnet.Internal; namespace MQTTnet.Server { @@ -14,10 +14,10 @@ namespace MQTTnet.Server private readonly AsyncLock _messagesLock = new AsyncLock(); private readonly Dictionary _messages = new Dictionary(); - private IMqttNetChildLogger _logger; + private IMqttNetLogger _logger; private IMqttServerOptions _options; - public Task Start(IMqttServerOptions options, IMqttNetChildLogger logger) + public Task Start(IMqttServerOptions options, IMqttNetLogger logger) { if (logger == null) throw new ArgumentNullException(nameof(logger)); _logger = logger.CreateChildLogger(nameof(MqttRetainedMessagesManager)); diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 4c5ab620..0fa6b4c9 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -17,13 +17,13 @@ namespace MQTTnet.Server { private readonly MqttServerEventDispatcher _eventDispatcher; private readonly ICollection _adapters; - private readonly IMqttNetChildLogger _logger; + private readonly IMqttNetLogger _logger; private MqttClientSessionsManager _clientSessionsManager; private IMqttRetainedMessagesManager _retainedMessagesManager; private CancellationTokenSource _cancellationTokenSource; - public MqttServer(IEnumerable adapters, IMqttNetChildLogger logger) + public MqttServer(IEnumerable adapters, IMqttNetLogger logger) { if (adapters == null) throw new ArgumentNullException(nameof(adapters)); _adapters = adapters.ToList(); diff --git a/Source/MQTTnet/Server/MqttServerEventDispatcher.cs b/Source/MQTTnet/Server/MqttServerEventDispatcher.cs index e6e608ac..3eb6b854 100644 --- a/Source/MQTTnet/Server/MqttServerEventDispatcher.cs +++ b/Source/MQTTnet/Server/MqttServerEventDispatcher.cs @@ -1,15 +1,15 @@ -using System; -using System.Threading.Tasks; -using MQTTnet.Client.Receiving; +using MQTTnet.Client.Receiving; using MQTTnet.Diagnostics; +using System; +using System.Threading.Tasks; namespace MQTTnet.Server { public class MqttServerEventDispatcher { - private readonly IMqttNetChildLogger _logger; + private readonly IMqttNetLogger _logger; - public MqttServerEventDispatcher(IMqttNetChildLogger logger) + public MqttServerEventDispatcher(IMqttNetLogger logger) { _logger = logger ?? throw new ArgumentNullException(nameof(logger)); } diff --git a/Source/MQTTnet/TopicFilter.cs b/Source/MQTTnet/TopicFilter.cs index 3b62e7e0..b086b48e 100644 --- a/Source/MQTTnet/TopicFilter.cs +++ b/Source/MQTTnet/TopicFilter.cs @@ -2,6 +2,7 @@ namespace MQTTnet { + // TODO: Consider renaming to "MqttTopicFilter" public class TopicFilter { public string Topic { get; set; } @@ -26,16 +27,16 @@ namespace MQTTnet public override string ToString() { return string.Concat( - "TopicFilter: [Topic=", + "TopicFilter: [Topic=", Topic, - "] [QualityOfServiceLevel=", + "] [QualityOfServiceLevel=", QualityOfServiceLevel, - "] [NoLocal=", - NoLocal, - "] [RetainAsPublished=", - RetainAsPublished, - "] [RetainHandling=", - RetainHandling, + "] [NoLocal=", + NoLocal, + "] [RetainAsPublished=", + RetainAsPublished, + "] [RetainHandling=", + RetainHandling, "]"); } } diff --git a/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs b/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs index cfc88d44..9be84bde 100644 --- a/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs @@ -10,7 +10,7 @@ namespace MQTTnet.Benchmarks public class LoggerBenchmark { private IMqttNetLogger _logger; - private IMqttNetChildLogger _childLogger; + private IMqttNetLogger _childLogger; private bool _useHandler; [GlobalSetup] diff --git a/Tests/MQTTnet.Benchmarks/Program.cs b/Tests/MQTTnet.Benchmarks/Program.cs index 280416d3..760cb52e 100644 --- a/Tests/MQTTnet.Benchmarks/Program.cs +++ b/Tests/MQTTnet.Benchmarks/Program.cs @@ -9,7 +9,7 @@ namespace MQTTnet.Benchmarks { public static void Main(string[] args) { - Console.WriteLine($"MQTTnet - BenchmarkApp.{TargetFrameworkInfoProvider.TargetFramework}"); + Console.WriteLine($"MQTTnet - BenchmarkApp.{TargetFrameworkProvider.TargetFramework}"); Console.WriteLine("1 = MessageProcessingBenchmark"); Console.WriteLine("2 = SerializerBenchmark"); Console.WriteLine("3 = LoggerBenchmark"); diff --git a/Tests/MQTTnet.Core.Tests/LowLevelMqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/LowLevelMqttClient_Tests.cs new file mode 100644 index 00000000..05440de3 --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/LowLevelMqttClient_Tests.cs @@ -0,0 +1,111 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client.Options; +using MQTTnet.LowLevelClient; +using MQTTnet.Packets; +using MQTTnet.Protocol; +using MQTTnet.Tests.Mockups; +using System.Collections.Generic; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace MQTTnet.Tests +{ + [TestClass] + public class LowLevelMqttClient_Tests + { + public TestContext TestContext { get; set; } + + [TestMethod] + public async Task Connect_And_Disconnect() + { + using (var testEnvironment = new TestEnvironment(TestContext)) + { + var server = await testEnvironment.StartServerAsync(); + + var factory = new MqttFactory(); + var lowLevelClient = factory.CreateLowLevelMqttClient(); + + await lowLevelClient.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build(), CancellationToken.None); + + await lowLevelClient.DisconnectAsync(CancellationToken.None); + } + } + + [TestMethod] + public async Task Authenticate() + { + using (var testEnvironment = new TestEnvironment(TestContext)) + { + var server = await testEnvironment.StartServerAsync(); + + var factory = new MqttFactory(); + var lowLevelClient = factory.CreateLowLevelMqttClient(); + + await lowLevelClient.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build(), CancellationToken.None); + + var receivedPacket = await Authenticate(lowLevelClient).ConfigureAwait(false); + + await lowLevelClient.DisconnectAsync(CancellationToken.None).ConfigureAwait(false); + + Assert.IsNotNull(receivedPacket); + Assert.AreEqual(MqttConnectReturnCode.ConnectionAccepted, receivedPacket.ReturnCode); + } + } + + [TestMethod] + public async Task Subscribe() + { + using (var testEnvironment = new TestEnvironment(TestContext)) + { + var server = await testEnvironment.StartServerAsync(); + + var factory = new MqttFactory(); + var lowLevelClient = factory.CreateLowLevelMqttClient(); + + await lowLevelClient.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build(), CancellationToken.None); + + await Authenticate(lowLevelClient).ConfigureAwait(false); + + var receivedPacket = await Subscribe(lowLevelClient, "a").ConfigureAwait(false); + + await lowLevelClient.DisconnectAsync(CancellationToken.None).ConfigureAwait(false); + + Assert.IsNotNull(receivedPacket); + Assert.AreEqual(MqttSubscribeReturnCode.SuccessMaximumQoS0, receivedPacket.ReturnCodes[0]); + } + } + + async Task Authenticate(ILowLevelMqttClient client) + { + await client.SendAsync(new MqttConnectPacket() + { + CleanSession = true, + ClientId = TestContext.TestName, + Username = "user", + Password = Encoding.UTF8.GetBytes("pass") + }, + CancellationToken.None).ConfigureAwait(false); + + return await client.ReceiveAsync(CancellationToken.None).ConfigureAwait(false) as MqttConnAckPacket; + } + + async Task Subscribe(ILowLevelMqttClient client, string topic) + { + await client.SendAsync(new MqttSubscribePacket + { + PacketIdentifier = 1, + TopicFilters = new List + { + new TopicFilter + { + Topic = topic + } + } + }, + CancellationToken.None).ConfigureAwait(false); + + return await client.ReceiveAsync(CancellationToken.None).ConfigureAwait(false) as MqttSubAckPacket; + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs index 2500a6f0..103142ee 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs @@ -1,6 +1,4 @@ -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.Disconnecting; @@ -10,10 +8,12 @@ using MQTTnet.Client.Publishing; using MQTTnet.Client.Receiving; using MQTTnet.Client.Subscribing; using MQTTnet.Client.Unsubscribing; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Tests.Mockups { - public class TestClientWrapper : IMqttClient + public sealed class TestClientWrapper : IMqttClient { public TestClientWrapper(IMqttClient implementation, TestContext testContext) { @@ -22,6 +22,7 @@ namespace MQTTnet.Tests.Mockups } public IMqttClient Implementation { get; } + public TestContext TestContext { get; } public bool IsConnected => Implementation.IsConnected; @@ -34,28 +35,32 @@ namespace MQTTnet.Tests.Mockups public Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken) { - switch (options) + if (TestContext != null) { - case MqttClientOptionsBuilder builder: - { - var existingClientId = builder.Build().ClientId; - if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName)) + switch (options) + { + case MqttClientOptionsBuilder builder: { - builder.WithClientId(TestContext.TestName + existingClientId); + var existingClientId = builder.Build().ClientId; + if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName)) + { + builder.WithClientId(TestContext.TestName + existingClientId); + } + + break; } - } - break; - case MqttClientOptions op: - { - var existingClientId = op.ClientId; - if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName)) + + case MqttClientOptions op: { - op.ClientId = TestContext.TestName + existingClientId; + var existingClientId = op.ClientId; + if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName)) + { + op.ClientId = TestContext.TestName + existingClientId; + } + + break; } - } - break; - default: - break; + } } return Implementation.ConnectAsync(options, cancellationToken); @@ -81,7 +86,7 @@ namespace MQTTnet.Tests.Mockups return Implementation.SendExtendedAuthenticationExchangeDataAsync(data, cancellationToken); } - public Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken) + public Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken) { return Implementation.SubscribeAsync(options, cancellationToken); } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index 7f2dcac1..be995343 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs @@ -1,27 +1,27 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading.Tasks; -using Microsoft.VisualStudio.TestTools.UnitTesting; +using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Client.Options; using MQTTnet.Diagnostics; using MQTTnet.Internal; using MQTTnet.Server; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; namespace MQTTnet.Tests.Mockups { - public class TestEnvironment : Disposable + public sealed class TestEnvironment : Disposable { - private readonly MqttFactory _mqttFactory = new MqttFactory(); - private readonly List _clients = new List(); - private readonly IMqttNetLogger _serverLogger = new MqttNetLogger("server"); - private readonly IMqttNetLogger _clientLogger = new MqttNetLogger("client"); + readonly MqttFactory _mqttFactory = new MqttFactory(); + readonly List _clients = new List(); + readonly IMqttNetLogger _serverLogger = new MqttNetLogger("server"); + readonly IMqttNetLogger _clientLogger = new MqttNetLogger("client"); - private readonly List _serverErrors = new List(); - private readonly List _clientErrors = new List(); + readonly List _serverErrors = new List(); + readonly List _clientErrors = new List(); - private readonly List _exceptions = new List(); + readonly List _exceptions = new List(); public IMqttServer Server { get; private set; } @@ -37,36 +37,42 @@ namespace MQTTnet.Tests.Mockups public TestContext TestContext { get; } + public TestEnvironment() : this(null) + { + } + public TestEnvironment(TestContext testContext) { + TestContext = testContext; + _serverLogger.LogMessagePublished += (s, e) => { - if (e.TraceMessage.Level == MqttNetLogLevel.Error) + if (e.LogMessage.Level == MqttNetLogLevel.Error) { lock (_serverErrors) { - _serverErrors.Add(e.TraceMessage.ToString()); + _serverErrors.Add(e.LogMessage.ToString()); } } }; _clientLogger.LogMessagePublished += (s, e) => { - lock (_clientErrors) + if (e.LogMessage.Level == MqttNetLogLevel.Error) { - if (e.TraceMessage.Level == MqttNetLogLevel.Error) + lock (_clientErrors) { - _clientErrors.Add(e.TraceMessage.ToString()); + _clientErrors.Add(e.LogMessage.ToString()); } } }; - TestContext = testContext; } public IMqttClient CreateClient() { var client = _mqttFactory.CreateMqttClient(_clientLogger); _clients.Add(client); + return new TestClientWrapper(client, TestContext); } @@ -90,15 +96,17 @@ namespace MQTTnet.Tests.Mockups public Task ConnectClientAsync() { - return ConnectClientAsync(new MqttClientOptionsBuilder() ); + return ConnectClientAsync(new MqttClientOptionsBuilder()); } public async Task ConnectClientAsync(MqttClientOptionsBuilder options) { if (options == null) throw new ArgumentNullException(nameof(options)); + options = options.WithTcpServer("localhost", ServerPort); + var client = CreateClient(); - await client.ConnectAsync(options.WithTcpServer("localhost", ServerPort).Build()); + await client.ConnectAsync(options.Build()); return client; } @@ -150,6 +158,7 @@ namespace MQTTnet.Tests.Mockups throw new Exception($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions)); } } + base.Dispose(disposing); } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestLogger.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestLogger.cs index 79607d97..46471f83 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestLogger.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestLogger.cs @@ -1,20 +1,15 @@ -using System; -using MQTTnet.Diagnostics; +using MQTTnet.Diagnostics; +using System; namespace MQTTnet.Tests.Mockups { - public class TestLogger : IMqttNetLogger, IMqttNetChildLogger + public class TestLogger : IMqttNetLogger { public event EventHandler LogMessagePublished; - IMqttNetChildLogger IMqttNetLogger.CreateChildLogger(string source) + public IMqttNetLogger CreateChildLogger(string source = null) { - return new MqttNetChildLogger(this, source); - } - - IMqttNetChildLogger IMqttNetChildLogger.CreateChildLogger(string source) - { - return new MqttNetChildLogger(this, source); + return new TestLogger(); } public void Verbose(string message, params object[] parameters) @@ -36,5 +31,10 @@ namespace MQTTnet.Tests.Mockups public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception) { } + + public void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception) + { + throw new NotImplementedException(); + } } } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapterFactory.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapterFactory.cs index ff47f710..95fbdfe2 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapterFactory.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapterFactory.cs @@ -12,8 +12,8 @@ namespace MQTTnet.Tests.Mockups { _adapter = adapter; } - - public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger) + + public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger) { return _adapter; } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs index f9901978..79c97393 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs @@ -1,16 +1,16 @@ -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.Client.Publishing; using MQTTnet.Client.Receiving; using MQTTnet.Server; using MQTTnet.Server.Status; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Tests.Mockups { - public class TestServerWrapper : IMqttServer + public sealed class TestServerWrapper : IMqttServer { public TestServerWrapper(IMqttServer implementation, TestContext testContext, TestEnvironment testEnvironment) { @@ -60,22 +60,29 @@ namespace MQTTnet.Tests.Mockups public Task StartAsync(IMqttServerOptions options) { - switch (options) + if (TestContext != 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; - default: - break; + switch (options) + { + 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; + } + } } return Implementation.StartAsync(options); diff --git a/Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs index eb7aba36..28f25118 100644 --- a/Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs @@ -17,22 +17,23 @@ namespace MQTTnet.Tests //This test compares //1. correct logID - string logId = "logId"; + var logId = "logId"; string invalidLogId = null; //2. if the total log calls are the same for global and local - int globalLogCount = 0; - int localLogCount = 0; + var globalLogCount = 0; + var localLogCount = 0; - MqttNetLogger logger = new MqttNetLogger(logId); + var logger = new MqttNetLogger(logId); //we have a theoretical bug here if a concurrent test is also logging var globalLog = new EventHandler((s, e) => { - if (logId != e.TraceMessage.LogId) + if (logId != e.LogMessage.LogId) { - invalidLogId = e.TraceMessage.LogId; + invalidLogId = e.LogMessage.LogId; } + Interlocked.Increment(ref globalLogCount); }); @@ -40,10 +41,11 @@ namespace MQTTnet.Tests logger.LogMessagePublished += (s, e) => { - if (logId != e.TraceMessage.LogId) + if (logId != e.LogMessage.LogId) { - invalidLogId = e.TraceMessage.LogId; + invalidLogId = e.LogMessage.LogId; } + Interlocked.Increment(ref localLogCount); }; diff --git a/Tests/MQTTnet.Core.Tests/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/Server_Tests.cs index 3ebdaa6d..7d81eca8 100644 --- a/Tests/MQTTnet.Core.Tests/Server_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server_Tests.cs @@ -1,11 +1,4 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Net.Sockets; -using System.Text; -using System.Threading; -using System.Threading.Tasks; -using Microsoft.VisualStudio.TestTools.UnitTesting; +using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Adapter; using MQTTnet.Client; using MQTTnet.Client.Connecting; @@ -17,6 +10,13 @@ using MQTTnet.Implementations; 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.Text; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Tests { @@ -54,7 +54,7 @@ namespace MQTTnet.Tests MqttQualityOfServiceLevel.AtMostOnce, "A/B/C", MqttQualityOfServiceLevel.AtMostOnce, - 1, + 1, TestContext); } @@ -1016,7 +1016,7 @@ namespace MQTTnet.Tests [TestMethod] public async Task Same_Client_Id_Connect_Disconnect_Event_Order() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = new TestEnvironment()) { var server = await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder()); @@ -1038,11 +1038,11 @@ namespace MQTTnet.Tests } }); - var clientOptions = new MqttClientOptionsBuilder() - .WithClientId("same_id"); + var clientOptionsBuilder = new MqttClientOptionsBuilder() + .WithClientId(Guid.NewGuid().ToString()); // c - var c1 = await testEnvironment.ConnectClientAsync(clientOptions); + var c1 = await testEnvironment.ConnectClientAsync(clientOptionsBuilder); await Task.Delay(500); @@ -1050,7 +1050,7 @@ namespace MQTTnet.Tests Assert.AreEqual("c", flow); // dc - var c2 = await testEnvironment.ConnectClientAsync(clientOptions); + var c2 = await testEnvironment.ConnectClientAsync(clientOptionsBuilder); c2.UseApplicationMessageReceivedHandler(_ => { @@ -1058,8 +1058,8 @@ namespace MQTTnet.Tests { events.Add("r"); } - }); + c2.SubscribeAsync("topic").Wait(); await Task.Delay(500); @@ -1080,7 +1080,7 @@ namespace MQTTnet.Tests Assert.AreEqual(false, c1.IsConnected); await c1.DisconnectAsync(); - Assert.AreEqual (false, c1.IsConnected); + Assert.AreEqual(false, c1.IsConnected); await Task.Delay(500); @@ -1141,7 +1141,7 @@ namespace MQTTnet.Tests await testEnvironment.ConnectClientAsync(); } } - + [TestMethod] public async Task Close_Idle_Connection() { @@ -1182,7 +1182,7 @@ namespace MQTTnet.Tests // forever. This is security related. var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); await PlatformAbstractionLayer.ConnectAsync(client, "localhost", testEnvironment.ServerPort); - + var buffer = Encoding.UTF8.GetBytes("Garbage"); client.Send(buffer, buffer.Length, SocketFlags.None); diff --git a/Tests/MQTTnet.TestApp.NetCore/Program.cs b/Tests/MQTTnet.TestApp.NetCore/Program.cs index 777c6c07..9b2cbeb9 100644 --- a/Tests/MQTTnet.TestApp.NetCore/Program.cs +++ b/Tests/MQTTnet.TestApp.NetCore/Program.cs @@ -17,7 +17,7 @@ namespace MQTTnet.TestApp.NetCore { public static void Main() { - Console.WriteLine($"MQTTnet - TestApp.{TargetFrameworkInfoProvider.TargetFramework}"); + Console.WriteLine($"MQTTnet - TestApp.{TargetFrameworkProvider.TargetFramework}"); Console.WriteLine("1 = Start client"); Console.WriteLine("2 = Start server"); Console.WriteLine("3 = Start performance test"); From 81b547f3868861fdbc5be37d2ebb6e553e2903f8 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 1 Apr 2020 23:15:08 +0200 Subject: [PATCH 2/5] Fix logging issues. --- .../Logging/MqttNetLoggerWrapper.cs | 4 ++-- Source/MQTTnet/Diagnostics/MqttNetLogger.cs | 7 ++++++- Source/MQTTnet/MqttFactory.cs | 2 +- Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs | 2 +- Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs | 2 +- Tests/MQTTnet.Core.Tests/MqttNetLogger_Tests.cs | 17 +++++++++++++++++ 6 files changed, 28 insertions(+), 6 deletions(-) diff --git a/Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs b/Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs index adaa3632..9c71e28a 100644 --- a/Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs +++ b/Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs @@ -7,7 +7,7 @@ namespace MQTTnet.Server.Logging { public class MqttNetLoggerWrapper : IMqttNetLogger { - private readonly ILogger _logger; + readonly ILogger _logger; public MqttNetLoggerWrapper(ILogger logger) { @@ -16,7 +16,7 @@ namespace MQTTnet.Server.Logging public event EventHandler LogMessagePublished; - public IMqttNetLogger CreateChildLogger(string source = null) + public IMqttNetLogger CreateChildLogger(string source) { return new MqttNetLogger(source); } diff --git a/Source/MQTTnet/Diagnostics/MqttNetLogger.cs b/Source/MQTTnet/Diagnostics/MqttNetLogger.cs index 71610d41..bfd4c426 100644 --- a/Source/MQTTnet/Diagnostics/MqttNetLogger.cs +++ b/Source/MQTTnet/Diagnostics/MqttNetLogger.cs @@ -9,7 +9,7 @@ namespace MQTTnet.Diagnostics readonly MqttNetLogger _parentLogger; - public MqttNetLogger(string source, string logId = null) + public MqttNetLogger(string source, string logId) { _source = source; _logId = logId; @@ -19,6 +19,11 @@ namespace MQTTnet.Diagnostics { } + public MqttNetLogger(string logId) + { + _logId = logId; + } + MqttNetLogger(MqttNetLogger parentLogger, string logId, string source) { _parentLogger = parentLogger ?? throw new ArgumentNullException(nameof(parentLogger)); diff --git a/Source/MQTTnet/MqttFactory.cs b/Source/MQTTnet/MqttFactory.cs index ed70fd51..ffc8f90f 100644 --- a/Source/MQTTnet/MqttFactory.cs +++ b/Source/MQTTnet/MqttFactory.cs @@ -13,7 +13,7 @@ namespace MQTTnet { IMqttClientAdapterFactory _clientAdapterFactory = new MqttClientAdapterFactory(); - public MqttFactory() : this(new MqttNetLogger(null, null)) + public MqttFactory() : this(new MqttNetLogger()) { } diff --git a/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs b/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs index 9be84bde..da772990 100644 --- a/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs @@ -16,7 +16,7 @@ namespace MQTTnet.Benchmarks [GlobalSetup] public void Setup() { - _logger = new MqttNetLogger("1"); + _logger = new MqttNetLogger(); _childLogger = _logger.CreateChildLogger("child"); MqttNetGlobalLogger.LogMessagePublished += OnLogMessagePublished; diff --git a/Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs index d586d601..5ec2f9f3 100644 --- a/Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs @@ -24,7 +24,7 @@ namespace MQTTnet.Tests //var globalLogCount = 0; var localLogCount = 0; - var logger = new MqttNetLogger(logId); + var logger = new MqttNetLogger(null, 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 diff --git a/Tests/MQTTnet.Core.Tests/MqttNetLogger_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttNetLogger_Tests.cs index 760d70fa..8a36ff35 100644 --- a/Tests/MQTTnet.Core.Tests/MqttNetLogger_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttNetLogger_Tests.cs @@ -46,5 +46,22 @@ namespace MQTTnet.Tests Assert.AreEqual(4, logMessagesCount); } + + [TestMethod] + public void Set_Custom_Log_ID() + { + var logger = new MqttNetLogger(null, "logId"); + var childLogger = logger.CreateChildLogger("Source1"); + + logger.LogMessagePublished += (s, e) => + { + Assert.AreEqual("logId", e.LogMessage.LogId); + }; + + childLogger.Verbose("Verbose"); + childLogger.Info("Info"); + childLogger.Warning(null, "Warning"); + childLogger.Error(null, "Error"); + } } } From ad128c7889d039b81c3748547e73998e2535bbbc Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 3 Apr 2020 22:52:55 +0200 Subject: [PATCH 3/5] Improve socket handling. --- .../Implementations/CrossPlatformSocket.cs | 229 ++++++++++++++++++ .../MQTTnet/Implementations/MqttTcpChannel.cs | 38 ++- .../Implementations/MqttTcpServerListener.cs | 14 +- .../PlatformAbstractionLayer.cs | 87 +------ Source/MQTTnet/Server/MqttClientConnection.cs | 28 +-- .../Server/MqttClientSessionsManager.cs | 11 +- .../Server/MqttClientSubscriptionsManager.cs | 18 +- .../Server/MqttServerEventDispatcher.cs | 82 ++++--- .../CrossPlatformSocket_Tests.cs | 74 ++++++ .../MqttTcpChannel_Tests.cs | 18 +- Tests/MQTTnet.Core.Tests/Server_Tests.cs | 33 +-- 11 files changed, 438 insertions(+), 194 deletions(-) create mode 100644 Source/MQTTnet/Implementations/CrossPlatformSocket.cs create mode 100644 Tests/MQTTnet.Core.Tests/CrossPlatformSocket_Tests.cs diff --git a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs new file mode 100644 index 00000000..4d7f6c97 --- /dev/null +++ b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs @@ -0,0 +1,229 @@ +using System; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; + +namespace MQTTnet.Implementations +{ + public sealed class CrossPlatformSocket : IDisposable + { + readonly Socket _socket; + + public CrossPlatformSocket(AddressFamily addressFamily) + { + _socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp); + } + + public CrossPlatformSocket() + { + // Having this contructor is important because avoiding the address family as parameter + // will make use of dual mode in the .net framework. + _socket = new Socket(SocketType.Stream, ProtocolType.Tcp); + } + + public CrossPlatformSocket(Socket socket) + { + _socket = socket ?? throw new ArgumentNullException(nameof(socket)); + } + + public bool NoDelay + { + get + { + return (int)_socket.GetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay) > 0; + } + + set + { + _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, value ? 1 : 0); + } + } + + public bool DualMode + { + get + { + return _socket.DualMode; + } + + set + { + _socket.DualMode = value; + } + } + + public int ReceiveBufferSize + { + get + { + return _socket.ReceiveBufferSize; + } + + set + { + _socket.ReceiveBufferSize = value; + } + } + + public int SendBufferSize + { + get + { + return _socket.SendBufferSize; + } + + set + { + _socket.SendBufferSize = value; + } + } + + public EndPoint RemoteEndPoint + { + get + { + return _socket.RemoteEndPoint; + } + } + + public bool ReuseAddress + { + get + { + return (int)_socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress) != 0; + } + + set + { + _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, value ? 1 : 0); + } + } + + public async Task AcceptAsync() + { + try + { +#if NET452 || NET461 + var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false); + return new CrossPlatformSocket(clientSocket); +#else + var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false); + return new CrossPlatformSocket(clientSocket); +#endif + } + catch (ObjectDisposedException) + { + // This will happen when _socket.EndAccept gets called by Task library but the socket is already disposed. + return null; + } + } + + public void Bind(EndPoint localEndPoint) + { + if (localEndPoint is null) throw new ArgumentNullException(nameof(localEndPoint)); + + _socket.Bind(localEndPoint); + } + + public void Listen(int connectionBacklog) + { + _socket.Listen(connectionBacklog); + } + + public async Task ConnectAsync(string host, int port, CancellationToken cancellationToken) + { + if (host is null) throw new ArgumentNullException(nameof(host)); + + try + { + // Workaround for: workaround for https://github.com/dotnet/corefx/issues/24430 + using (cancellationToken.Register(() => _socket.Dispose())) + { + cancellationToken.ThrowIfCancellationRequested(); + +#if NET452 || NET461 + await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, host, port, null).ConfigureAwait(false); +#else + await _socket.ConnectAsync(host, port).ConfigureAwait(false); +#endif + } + } + catch (ObjectDisposedException) + { + // This will happen when _socket.EndConnect gets called by Task library but the socket is already disposed. + } + } + + public async Task SendAsync(ArraySegment buffer, SocketFlags socketFlags) + { + try + { +#if NET452 || NET461 + await Task.Factory.FromAsync(SocketWrapper.BeginSend, _socket.EndSend, new SocketWrapper(_socket, buffer, socketFlags)).ConfigureAwait(false); +#else + await _socket.SendAsync(buffer, socketFlags).ConfigureAwait(false); +#endif + } + catch (ObjectDisposedException) + { + // This will happen when _socket.EndConnect gets called by Task library but the socket is already disposed. + } + } + + public async Task ReceiveAsync(ArraySegment buffer, SocketFlags socketFlags) + { + try + { +#if NET452 || NET461 + return await Task.Factory.FromAsync(SocketWrapper.BeginReceive, _socket.EndReceive, new SocketWrapper(_socket, buffer, socketFlags)).ConfigureAwait(false); +#else + return await _socket.ReceiveAsync(buffer, socketFlags).ConfigureAwait(false); +#endif + } + catch (ObjectDisposedException) + { + // This will happen when _socket.EndReceive gets called by Task library but the socket is already disposed. + return -1; + } + } + + public NetworkStream GetStream() + { + return new NetworkStream(_socket, true); + } + + public void Dispose() + { + _socket?.Dispose(); + } + +#if NET452 || NET461 + class SocketWrapper + { + readonly Socket _socket; + readonly ArraySegment _buffer; + readonly SocketFlags _socketFlags; + + public SocketWrapper(Socket socket, ArraySegment buffer, SocketFlags socketFlags) + { + _socket = socket; + _buffer = buffer; + _socketFlags = socketFlags; + } + + public static IAsyncResult BeginSend(AsyncCallback callback, object state) + { + var socketWrapper = (SocketWrapper)state; + return socketWrapper._socket.BeginSend(socketWrapper._buffer.Array, socketWrapper._buffer.Offset, socketWrapper._buffer.Count, socketWrapper._socketFlags, callback, state); + } + + public static IAsyncResult BeginReceive(AsyncCallback callback, object state) + { + var socketWrapper = (SocketWrapper)state; + return socketWrapper._socket.BeginReceive(socketWrapper._buffer.Array, socketWrapper._buffer.Offset, socketWrapper._buffer.Count, socketWrapper._socketFlags, callback, state); + } + } +#endif + } +} diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index 71050e1e..021a7d49 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -46,15 +46,15 @@ namespace MQTTnet.Implementations public async Task ConnectAsync(CancellationToken cancellationToken) { - Socket socket; + CrossPlatformSocket socket; if (_options.AddressFamily == AddressFamily.Unspecified) { - socket = new Socket(SocketType.Stream, ProtocolType.Tcp); + socket = new CrossPlatformSocket(); } else { - socket = new Socket(_options.AddressFamily, SocketType.Stream, ProtocolType.Tcp); + socket = new CrossPlatformSocket(_options.AddressFamily); } socket.ReceiveBufferSize = _options.BufferSize; @@ -69,20 +69,24 @@ namespace MQTTnet.Implementations socket.DualMode = _options.DualMode.Value; } - // Workaround for: workaround for https://github.com/dotnet/corefx/issues/24430 - using (cancellationToken.Register(() => socket.Dispose())) - { - await PlatformAbstractionLayer.ConnectAsync(socket, _options.Server, _options.GetPort()).ConfigureAwait(false); - } + await socket.ConnectAsync(_options.Server, _options.GetPort(), cancellationToken).ConfigureAwait(false); - var networkStream = new NetworkStream(socket, true); + var networkStream = socket.GetStream(); if (_options.TlsOptions.UseTls) { var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback); - _stream = sslStream; + try + { + await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, !_options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false); + } + catch + { + sslStream.Dispose(); + throw; + } - await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, !_options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false); + _stream = sslStream; } else { @@ -107,17 +111,14 @@ namespace MQTTnet.Implementations // Workaround for: https://github.com/dotnet/corefx/issues/24430 using (cancellationToken.Register(Dispose)) { - if (cancellationToken.IsCancellationRequested) - { - return 0; - } + cancellationToken.ThrowIfCancellationRequested(); return await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); } } catch (ObjectDisposedException) { - return 0; + return -1; } catch (IOException exception) { @@ -139,10 +140,7 @@ namespace MQTTnet.Implementations // Workaround for: https://github.com/dotnet/corefx/issues/24430 using (cancellationToken.Register(Dispose)) { - if (cancellationToken.IsCancellationRequested) - { - return; - } + cancellationToken.ThrowIfCancellationRequested(); await _stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); } diff --git a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs index 84cd7b96..83ef80fd 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs @@ -23,7 +23,7 @@ namespace MQTTnet.Implementations readonly MqttServerTlsTcpEndpointOptions _tlsOptions; readonly X509Certificate2 _tlsCertificate; - private Socket _socket; + private CrossPlatformSocket _socket; private IPEndPoint _localEndPoint; public MqttTcpServerListener( @@ -59,18 +59,18 @@ namespace MQTTnet.Implementations _logger.Info($"Starting TCP listener for {_localEndPoint} TLS={_tlsCertificate != null}."); - _socket = new Socket(_addressFamily, SocketType.Stream, ProtocolType.Tcp); + _socket = new CrossPlatformSocket(_addressFamily); // Usage of socket options is described here: https://docs.microsoft.com/en-us/dotnet/api/system.net.sockets.socket.setsocketoption?view=netcore-2.2 if (_options.ReuseAddress) { - _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true); + _socket.ReuseAddress = true; } if (_options.NoDelay) { - _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, true); + _socket.NoDelay = true; } _socket.Bind(_localEndPoint); @@ -107,7 +107,7 @@ namespace MQTTnet.Implementations { try { - var clientSocket = await PlatformAbstractionLayer.AcceptAsync(_socket).ConfigureAwait(false); + var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false); if (clientSocket == null) { continue; @@ -135,7 +135,7 @@ namespace MQTTnet.Implementations } } - async Task TryHandleClientConnectionAsync(Socket clientSocket) + async Task TryHandleClientConnectionAsync(CrossPlatformSocket clientSocket) { Stream stream = null; string remoteEndPoint = null; @@ -151,7 +151,7 @@ namespace MQTTnet.Implementations clientSocket.NoDelay = _options.NoDelay; - stream = new NetworkStream(clientSocket, true); + stream = clientSocket.GetStream(); X509Certificate2 clientCertificate = null; diff --git a/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs b/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs index 80c08903..0b683dc5 100644 --- a/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs +++ b/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs @@ -1,94 +1,9 @@ -using System; -using System.Net; -using System.Net.Sockets; -using System.Threading.Tasks; +using System.Threading.Tasks; namespace MQTTnet.Implementations { public static class PlatformAbstractionLayer { - // TODO: Consider creating primitives like "MqttNetSocket" which will wrap all required methods and do the platform stuff. - public static async Task AcceptAsync(Socket socket) - { -#if NET452 || NET461 - try - { - return await Task.Factory.FromAsync(socket.BeginAccept, socket.EndAccept, null).ConfigureAwait(false); - } - catch (ObjectDisposedException) - { - return null; - } -#else - return await socket.AcceptAsync().ConfigureAwait(false); -#endif - } - - - public static Task ConnectAsync(Socket socket, IPAddress ip, int port) - { -#if NET452 || NET461 - return Task.Factory.FromAsync(socket.BeginConnect, socket.EndConnect, ip, port, null); -#else - return socket.ConnectAsync(ip, port); -#endif - } - - public static Task ConnectAsync(Socket socket, string host, int port) - { -#if NET452 || NET461 - return Task.Factory.FromAsync(socket.BeginConnect, socket.EndConnect, host, port, null); -#else - return socket.ConnectAsync(host, port); -#endif - } - -#if NET452 || NET461 - public class SocketWrapper - { - private readonly Socket _socket; - private readonly ArraySegment _buffer; - private readonly SocketFlags _socketFlags; - - public SocketWrapper(Socket socket, ArraySegment buffer, SocketFlags socketFlags) - { - _socket = socket; - _buffer = buffer; - _socketFlags = socketFlags; - } - - public static IAsyncResult BeginSend(AsyncCallback callback, object state) - { - var real = (SocketWrapper)state; - return real._socket.BeginSend(real._buffer.Array, real._buffer.Offset, real._buffer.Count, real._socketFlags, callback, state); - } - - public static IAsyncResult BeginReceive(AsyncCallback callback, object state) - { - var real = (SocketWrapper)state; - return real._socket.BeginReceive(real._buffer.Array, real._buffer.Offset, real._buffer.Count, real._socketFlags, callback, state); - } - } -#endif - - public static Task SendAsync(Socket socket, ArraySegment buffer, SocketFlags socketFlags) - { -#if NET452 || NET461 - return Task.Factory.FromAsync(SocketWrapper.BeginSend, socket.EndSend, new SocketWrapper(socket, buffer, socketFlags)); -#else - return socket.SendAsync(buffer, socketFlags); -#endif - } - - public static Task ReceiveAsync(Socket socket, ArraySegment buffer, SocketFlags socketFlags) - { -#if NET452 || NET461 - return Task.Factory.FromAsync(SocketWrapper.BeginReceive, socket.EndReceive, new SocketWrapper(socket, buffer, socketFlags)); -#else - return socket.ReceiveAsync(buffer, socketFlags); -#endif - } - public static Task CompletedTask { get diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index 54475d08..70beba82 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -15,7 +15,7 @@ using System.Threading.Tasks; namespace MQTTnet.Server { - public class MqttClientConnection : IDisposable + public sealed class MqttClientConnection : IDisposable { private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); @@ -124,7 +124,7 @@ namespace MQTTnet.Server return _packageReceiverTask; } - private async Task RunInternalAsync(MqttConnectionValidatorContext connectionValidatorContext) + async Task RunInternalAsync(MqttConnectionValidatorContext connectionValidatorContext) { var disconnectType = MqttClientDisconnectType.NotClean; try @@ -251,12 +251,12 @@ namespace MQTTnet.Server return disconnectType; } - private void StopInternal() + void StopInternal() { _cancellationToken.Cancel(false); } - private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection topicFilters) + async Task EnqueueSubscribedRetainedMessagesAsync(ICollection topicFilters) { var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false); foreach (var applicationMessage in retainedMessages) @@ -265,7 +265,7 @@ namespace MQTTnet.Server } } - private async Task HandleIncomingSubscribePacketAsync(MqttSubscribePacket subscribePacket) + async Task HandleIncomingSubscribePacketAsync(MqttSubscribePacket subscribePacket) { // TODO: Let the channel adapter create the packet. var subscribeResult = await Session.SubscriptionsManager.SubscribeAsync(subscribePacket, ConnectPacket).ConfigureAwait(false); @@ -281,14 +281,14 @@ namespace MQTTnet.Server await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false); } - private async Task HandleIncomingUnsubscribePacketAsync(MqttUnsubscribePacket unsubscribePacket) + async Task HandleIncomingUnsubscribePacketAsync(MqttUnsubscribePacket unsubscribePacket) { // TODO: Let the channel adapter create the packet. var unsubscribeResult = await Session.SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false); await SendAsync(unsubscribeResult).ConfigureAwait(false); } - private Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket) + Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket) { Interlocked.Increment(ref _sentApplicationMessagesCount); @@ -313,7 +313,7 @@ namespace MQTTnet.Server } } - private Task HandleIncomingPublishPacketWithQoS0Async(MqttPublishPacket publishPacket) + Task HandleIncomingPublishPacketWithQoS0Async(MqttPublishPacket publishPacket) { var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket); @@ -322,7 +322,7 @@ namespace MQTTnet.Server return Task.FromResult(0); } - private Task HandleIncomingPublishPacketWithQoS1Async(MqttPublishPacket publishPacket) + Task HandleIncomingPublishPacketWithQoS1Async(MqttPublishPacket publishPacket) { var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket); _sessionsManager.DispatchApplicationMessage(applicationMessage, this); @@ -331,7 +331,7 @@ namespace MQTTnet.Server return SendAsync(pubAckPacket); } - private Task HandleIncomingPublishPacketWithQoS2Async(MqttPublishPacket publishPacket) + Task HandleIncomingPublishPacketWithQoS2Async(MqttPublishPacket publishPacket) { var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket); _sessionsManager.DispatchApplicationMessage(applicationMessage, this); @@ -345,7 +345,7 @@ namespace MQTTnet.Server return SendAsync(pubRecPacket); } - private async Task SendPendingPacketsAsync(CancellationToken cancellationToken) + async Task SendPendingPacketsAsync(CancellationToken cancellationToken) { MqttQueuedApplicationMessage queuedApplicationMessage = null; MqttPublishPacket publishPacket = null; @@ -459,7 +459,7 @@ namespace MQTTnet.Server } } - private async Task SendAsync(MqttBasePacket packet) + async Task SendAsync(MqttBasePacket packet) { await _channelAdapter.SendPacketAsync(packet, _serverOptions.DefaultCommunicationTimeout, _cancellationToken.Token).ConfigureAwait(false); @@ -471,12 +471,12 @@ namespace MQTTnet.Server } } - private void OnAdapterReadingPacketCompleted() + void OnAdapterReadingPacketCompleted() { _keepAliveMonitor?.Resume(); } - private void OnAdapterReadingPacketStarted() + void OnAdapterReadingPacketStarted() { _keepAliveMonitor?.Pause(); } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 285f63e9..08568d81 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -207,7 +207,7 @@ namespace MQTTnet.Server applicationMessage = interceptorContext.ApplicationMessage; } - await _eventDispatcher.HandleApplicationMessageReceivedAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false); + await _eventDispatcher.SafeNotifyApplicationMessageReceivedAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false); if (applicationMessage.Retain) { @@ -237,7 +237,7 @@ namespace MQTTnet.Server string clientId = null; var clientWasConnected = true; - MqttConnectPacket connectPacket = null; + MqttConnectPacket connectPacket; try { @@ -259,8 +259,6 @@ namespace MQTTnet.Server var connectionValidatorContext = await ValidateConnectionAsync(connectPacket, channelAdapter).ConfigureAwait(false); - clientId = connectPacket.ClientId; - if (connectionValidatorContext.ReasonCode != MqttConnectReasonCode.Success) { clientWasConnected = false; @@ -272,9 +270,10 @@ namespace MQTTnet.Server return; } + clientId = connectPacket.ClientId; var connection = await CreateConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false); - await _eventDispatcher.HandleClientConnectedAsync(clientId).ConfigureAwait(false); + await _eventDispatcher.SafeNotifyClientConnectedAsync(clientId).ConfigureAwait(false); disconnectType = await connection.RunAsync(connectionValidatorContext).ConfigureAwait(false); } @@ -303,7 +302,7 @@ namespace MQTTnet.Server if (clientId != null) { - await _eventDispatcher.TryHandleClientDisconnectedAsync(clientId, disconnectType).ConfigureAwait(false); + await _eventDispatcher.SafeNotifyClientDisconnectedAsync(clientId, disconnectType).ConfigureAwait(false); } } } diff --git a/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs index deeadf40..59e9f349 100644 --- a/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs @@ -1,9 +1,9 @@ -using System; +using MQTTnet.Packets; +using MQTTnet.Protocol; +using System; using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; -using MQTTnet.Packets; -using MQTTnet.Protocol; namespace MQTTnet.Server { @@ -67,7 +67,7 @@ namespace MQTTnet.Server _subscriptions[finalTopicFilter.Topic] = finalTopicFilter; } - await _eventDispatcher.HandleClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false); + await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false); } } @@ -83,7 +83,7 @@ namespace MQTTnet.Server var interceptorContext = await InterceptSubscribeAsync(topicFilter).ConfigureAwait(false); if (!interceptorContext.AcceptSubscription) { - continue; + continue; } if (interceptorContext.AcceptSubscription) @@ -93,7 +93,7 @@ namespace MQTTnet.Server _subscriptions[topicFilter.Topic] = topicFilter; } - await _eventDispatcher.HandleClientSubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false); + await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false); } } } @@ -131,9 +131,9 @@ namespace MQTTnet.Server foreach (var topicFilter in unsubscribePacket.TopicFilters) { - await _eventDispatcher.HandleClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false); + await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false); } - + return unsubAckPacket; } @@ -152,7 +152,7 @@ namespace MQTTnet.Server lock (_subscriptions) { _subscriptions.Remove(topicFilter); - } + } } } diff --git a/Source/MQTTnet/Server/MqttServerEventDispatcher.cs b/Source/MQTTnet/Server/MqttServerEventDispatcher.cs index 3eb6b854..3bb37681 100644 --- a/Source/MQTTnet/Server/MqttServerEventDispatcher.cs +++ b/Source/MQTTnet/Server/MqttServerEventDispatcher.cs @@ -7,7 +7,7 @@ namespace MQTTnet.Server { public class MqttServerEventDispatcher { - private readonly IMqttNetLogger _logger; + readonly IMqttNetLogger _logger; public MqttServerEventDispatcher(IMqttNetLogger logger) { @@ -24,18 +24,25 @@ namespace MQTTnet.Server public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; } - public Task HandleClientConnectedAsync(string clientId) + public async Task SafeNotifyClientConnectedAsync(string clientId) { - var handler = ClientConnectedHandler; - if (handler == null) + try { - return Task.FromResult(0); - } + var handler = ClientConnectedHandler; + if (handler == null) + { + return; + } - return handler.HandleClientConnectedAsync(new MqttServerClientConnectedEventArgs(clientId)); + await handler.HandleClientConnectedAsync(new MqttServerClientConnectedEventArgs(clientId)).ConfigureAwait(false); + } + catch (Exception exception) + { + _logger.Error(exception, "Error while handling custom 'ClientConnected' event."); + } } - public async Task TryHandleClientDisconnectedAsync(string clientId, MqttClientDisconnectType disconnectType) + public async Task SafeNotifyClientDisconnectedAsync(string clientId, MqttClientDisconnectType disconnectType) { try { @@ -49,41 +56,62 @@ namespace MQTTnet.Server } catch (Exception exception) { - _logger.Error(exception, "Error while handling 'ClientDisconnected' event."); + _logger.Error(exception, "Error while handling custom 'ClientDisconnected' event."); } } - public Task HandleClientSubscribedTopicAsync(string clientId, TopicFilter topicFilter) + public async Task SafeNotifyClientSubscribedTopicAsync(string clientId, TopicFilter topicFilter) { - var handler = ClientSubscribedTopicHandler; - if (handler == null) + try { - return Task.FromResult(0); - } + var handler = ClientSubscribedTopicHandler; + if (handler == null) + { + return; + } - return handler.HandleClientSubscribedTopicAsync(new MqttServerClientSubscribedTopicEventArgs(clientId, topicFilter)); + await handler.HandleClientSubscribedTopicAsync(new MqttServerClientSubscribedTopicEventArgs(clientId, topicFilter)).ConfigureAwait(false); + } + catch (Exception exception) + { + _logger.Error(exception, "Error while handling custom 'ClientSubscribedTopic' event."); + } } - public Task HandleClientUnsubscribedTopicAsync(string clientId, string topicFilter) + public async Task SafeNotifyClientUnsubscribedTopicAsync(string clientId, string topicFilter) { - var handler = ClientUnsubscribedTopicHandler; - if (handler == null) + try { - return Task.FromResult(0); - } + var handler = ClientUnsubscribedTopicHandler; + if (handler == null) + { + return; + } - return handler.HandleClientUnsubscribedTopicAsync(new MqttServerClientUnsubscribedTopicEventArgs(clientId, topicFilter)); + await handler.HandleClientUnsubscribedTopicAsync(new MqttServerClientUnsubscribedTopicEventArgs(clientId, topicFilter)).ConfigureAwait(false); + } + catch (Exception exception) + { + _logger.Error(exception, "Error while handling custom 'ClientUnsubscribedTopic' event."); + } } - public Task HandleApplicationMessageReceivedAsync(string senderClientId, MqttApplicationMessage applicationMessage) + public async Task SafeNotifyApplicationMessageReceivedAsync(string senderClientId, MqttApplicationMessage applicationMessage) { - var handler = ApplicationMessageReceivedHandler; - if (handler == null) + try { - return Task.FromResult(0); - } + var handler = ApplicationMessageReceivedHandler; + if (handler == null) + { + return; + } - return handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)); + await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)).ConfigureAwait(false); ; + } + catch (Exception exception) + { + _logger.Error(exception, "Error while handling custom 'ApplicationMessageReceived' event."); + } } } } diff --git a/Tests/MQTTnet.Core.Tests/CrossPlatformSocket_Tests.cs b/Tests/MQTTnet.Core.Tests/CrossPlatformSocket_Tests.cs new file mode 100644 index 00000000..3e3455e0 --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/CrossPlatformSocket_Tests.cs @@ -0,0 +1,74 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Implementations; +using System; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace MQTTnet.Tests +{ + [TestClass] + public class CrossPlatformSocket_Tests + { + [TestMethod] + public async Task Connect_Send_Receive() + { + var crossPlatformSocket = new CrossPlatformSocket(); + await crossPlatformSocket.ConnectAsync("www.google.de", 80, CancellationToken.None); + + var requestBuffer = Encoding.UTF8.GetBytes("GET / HTTP/1.1\r\nHost: www.google.de\r\n\r\n"); + await crossPlatformSocket.SendAsync(new ArraySegment(requestBuffer), System.Net.Sockets.SocketFlags.None); + + var buffer = new byte[1024]; + var length = await crossPlatformSocket.ReceiveAsync(new ArraySegment(buffer), System.Net.Sockets.SocketFlags.None); + crossPlatformSocket.Dispose(); + + var responseText = Encoding.UTF8.GetString(buffer, 0, length); + + Assert.IsTrue(responseText.Contains("HTTP/1.1 200 OK")); + } + + [TestMethod] + public async Task Try_Connect_Invalid_Host() + { + var crossPlatformSocket = new CrossPlatformSocket(); + + var cancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(3)); + cancellationToken.Token.Register(() => crossPlatformSocket.Dispose()); + + await crossPlatformSocket.ConnectAsync("www.google.de", 1234, CancellationToken.None); + } + + //[TestMethod] + //public async Task Use_Disconnected_Socket() + //{ + // var crossPlatformSocket = new CrossPlatformSocket(); + + // await crossPlatformSocket.ConnectAsync("www.google.de", 80); + + // var requestBuffer = Encoding.UTF8.GetBytes("GET /wrong_uri HTTP/1.1\r\nConnection: close\r\n\r\n"); + // await crossPlatformSocket.SendAsync(new ArraySegment(requestBuffer), System.Net.Sockets.SocketFlags.None); + + // var buffer = new byte[64000]; + // var length = await crossPlatformSocket.ReceiveAsync(new ArraySegment(buffer), System.Net.Sockets.SocketFlags.None); + + // await Task.Delay(500); + + // await crossPlatformSocket.SendAsync(new ArraySegment(requestBuffer), System.Net.Sockets.SocketFlags.None); + //} + + [TestMethod] + public async Task Set_Options() + { + var crossPlatformSocket = new CrossPlatformSocket(); + + Assert.IsFalse(crossPlatformSocket.ReuseAddress); + crossPlatformSocket.ReuseAddress = true; + Assert.IsTrue(crossPlatformSocket.ReuseAddress); + + Assert.IsFalse(crossPlatformSocket.NoDelay); + crossPlatformSocket.NoDelay = true; + Assert.IsTrue(crossPlatformSocket.NoDelay); + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/MqttTcpChannel_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttTcpChannel_Tests.cs index a4b0ca7c..6e78decb 100644 --- a/Tests/MQTTnet.Core.Tests/MqttTcpChannel_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttTcpChannel_Tests.cs @@ -1,10 +1,10 @@ -using System; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Implementations; +using System; using System.Net; using System.Net.Sockets; using System.Threading; using System.Threading.Tasks; -using Microsoft.VisualStudio.TestTools.UnitTesting; -using MQTTnet.Implementations; namespace MQTTnet.Tests { @@ -15,7 +15,7 @@ namespace MQTTnet.Tests public async Task Dispose_Channel_While_Used() { var ct = new CancellationTokenSource(); - var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); + var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork); try { @@ -28,18 +28,18 @@ namespace MQTTnet.Tests { while (!ct.IsCancellationRequested) { - var client = await PlatformAbstractionLayer.AcceptAsync(serverSocket); + var client = await serverSocket.AcceptAsync(); var data = new byte[] { 128 }; - await PlatformAbstractionLayer.SendAsync(client, new ArraySegment(data), SocketFlags.None); + await client.SendAsync(new ArraySegment(data), SocketFlags.None); } }, ct.Token); - var clientSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); - await PlatformAbstractionLayer.ConnectAsync(clientSocket, IPAddress.Loopback, 50001); + var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork); + await clientSocket.ConnectAsync("localhost", 50001, CancellationToken.None); await Task.Delay(100, ct.Token); - var tcpChannel = new MqttTcpChannel(new NetworkStream(clientSocket, true), "test", null); + var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), "test", null); var buffer = new byte[1]; await tcpChannel.ReadAsync(buffer, 0, 1, ct.Token); diff --git a/Tests/MQTTnet.Core.Tests/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/Server_Tests.cs index 8121f410..e40eba05 100644 --- a/Tests/MQTTnet.Core.Tests/Server_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server_Tests.cs @@ -904,14 +904,13 @@ namespace MQTTnet.Tests await testEnvironment.StartServerAsync(serverOptions); - var connectingFailedException = await Assert.ThrowsExceptionAsync(() => testEnvironment.ConnectClientAsync()); Assert.AreEqual(MqttClientConnectResultCode.NotAuthorized, connectingFailedException.ResultCode); } } + Dictionary _connected; - private Dictionary _connected; private void ConnectionValidationHandler(MqttConnectionValidatorContext eventArgs) { if (_connected.ContainsKey(eventArgs.ClientId)) @@ -919,6 +918,7 @@ namespace MQTTnet.Tests eventArgs.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; return; } + _connected[eventArgs.ClientId] = true; eventArgs.ReasonCode = MqttConnectReasonCode.Success; return; @@ -1053,6 +1053,12 @@ namespace MQTTnet.Tests // Connect client with same client ID. Should disconnect existing client. var c2 = await testEnvironment.ConnectClientAsync(clientOptionsBuilder); + await Task.Delay(500); + + flow = string.Join(string.Empty, events); + + Assert.AreEqual("cdc", flow); + c2.UseApplicationMessageReceivedHandler(_ => { lock (events) @@ -1061,15 +1067,10 @@ namespace MQTTnet.Tests } }); - c2.SubscribeAsync("topic").Wait(); - - await Task.Delay(500); - - flow = string.Join(string.Empty, events); - Assert.AreEqual("cdc", flow); + await c2.SubscribeAsync("topic"); // r - c2.PublishAsync("topic").Wait(); + await c2.PublishAsync("topic"); await Task.Delay(500); @@ -1149,15 +1150,15 @@ namespace MQTTnet.Tests { await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder().WithDefaultCommunicationTimeout(TimeSpan.FromSeconds(1))); - var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); - await PlatformAbstractionLayer.ConnectAsync(client, "localhost", testEnvironment.ServerPort); + var client = new CrossPlatformSocket(AddressFamily.InterNetwork); + await client.ConnectAsync("localhost", testEnvironment.ServerPort, CancellationToken.None); // Don't send anything. The server should close the connection. await Task.Delay(TimeSpan.FromSeconds(3)); try { - var receivedBytes = await PlatformAbstractionLayer.ReceiveAsync(client, new ArraySegment(new byte[10]), SocketFlags.Partial); + var receivedBytes = await client.ReceiveAsync(new ArraySegment(new byte[10]), SocketFlags.Partial); if (receivedBytes == 0) { return; @@ -1180,17 +1181,17 @@ namespace MQTTnet.Tests // Send an invalid packet and ensure that the server will close the connection and stay in a waiting state // forever. This is security related. - var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); - await PlatformAbstractionLayer.ConnectAsync(client, "localhost", testEnvironment.ServerPort); + var client = new CrossPlatformSocket(AddressFamily.InterNetwork); + await client.ConnectAsync("localhost", testEnvironment.ServerPort, CancellationToken.None); var buffer = Encoding.UTF8.GetBytes("Garbage"); - client.Send(buffer, buffer.Length, SocketFlags.None); + await client.SendAsync(new ArraySegment(buffer), SocketFlags.None); await Task.Delay(TimeSpan.FromSeconds(3)); try { - var receivedBytes = await PlatformAbstractionLayer.ReceiveAsync(client, new ArraySegment(new byte[10]), SocketFlags.Partial); + var receivedBytes = await client.ReceiveAsync(new ArraySegment(new byte[10]), SocketFlags.Partial); if (receivedBytes == 0) { return; From 7a24970035362040437a9e62abfd9c9aa3eb22e6 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 3 Apr 2020 23:08:54 +0200 Subject: [PATCH 4/5] Fix UnitTests. --- .../Extensions/UserPropertyExtension.cs | 18 +----------------- .../MqttApplicationMessage_Tests.cs | 11 +++++------ 2 files changed, 6 insertions(+), 23 deletions(-) diff --git a/Source/MQTTnet/Extensions/UserPropertyExtension.cs b/Source/MQTTnet/Extensions/UserPropertyExtension.cs index 5bb28cf9..f861c32d 100644 --- a/Source/MQTTnet/Extensions/UserPropertyExtension.cs +++ b/Source/MQTTnet/Extensions/UserPropertyExtension.cs @@ -1,32 +1,16 @@ using System; -using System.ComponentModel; using System.Linq; namespace MQTTnet.Extensions { public static class UserPropertyExtension { - public static string GetUserProperty(this MqttApplicationMessage message, string propertyName, StringComparison comparisonType = StringComparison.OrdinalIgnoreCase) + public static string GetUserProperty(this MqttApplicationMessage message, string propertyName, StringComparison comparisonType = StringComparison.Ordinal) { if (message == null) throw new ArgumentNullException(nameof(message)); if (propertyName == null) throw new ArgumentNullException(nameof(propertyName)); return message.UserProperties?.SingleOrDefault(up => up.Name.Equals(propertyName, comparisonType))?.Value; } - - public static T GetUserProperty(this MqttApplicationMessage message, string propertyName, StringComparison comparisonType = StringComparison.OrdinalIgnoreCase) - { - var value = GetUserProperty(message, propertyName, comparisonType); - - var typeDescriptor = TypeDescriptor.GetConverter(typeof(T)); - try - { - return (T) typeDescriptor.ConvertFromString(value); - } - catch (Exception ex) - { - throw new InvalidOperationException($"Cannot convert value({value ?? "null"}) of UserProperty({propertyName}) to {typeof(T).FullName}.", ex); - } - } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttApplicationMessage_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttApplicationMessage_Tests.cs index 6b791dc3..05d09072 100644 --- a/Tests/MQTTnet.Core.Tests/MqttApplicationMessage_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttApplicationMessage_Tests.cs @@ -1,8 +1,7 @@ -using System; -using System.Collections.Generic; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Extensions; using MQTTnet.Packets; +using System.Collections.Generic; namespace MQTTnet.Tests { @@ -23,11 +22,11 @@ namespace MQTTnet.Tests }; Assert.AreEqual("bar", message.GetUserProperty("foo")); - Assert.AreEqual(1011, message.GetUserProperty("value")); - Assert.AreEqual("insensitive", message.GetUserProperty("case")); + //Assert.AreEqual(1011, message.GetUserProperty("value")); + Assert.AreEqual(null, message.GetUserProperty("case")); Assert.AreEqual(null, message.GetUserProperty("nonExists")); - Assert.AreEqual(null, message.GetUserProperty("nonExists")); - Assert.ThrowsException(() => message.GetUserProperty("nonExists")); + //Assert.AreEqual(null, message.GetUserProperty("nonExists")); + //Assert.ThrowsException(() => message.GetUserProperty("nonExists")); } } } From d464bfb7c57f4b847d64955ec11905ace407bdc3 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 4 Apr 2020 01:01:37 +0200 Subject: [PATCH 5/5] Fix connect and disconnect event flow. --- Source/MQTTnet/Server/MqttClientConnection.cs | 65 ++++++++------ .../Server/MqttClientKeepAliveMonitor.cs | 12 +-- .../Server/MqttClientSessionsManager.cs | 84 +++++++++++-------- Source/MQTTnet/Server/MqttServer.cs | 14 ++-- 4 files changed, 98 insertions(+), 77 deletions(-) diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index 70beba82..a163b81d 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -3,6 +3,7 @@ using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Formatter; +using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.PacketDispatcher; using MQTTnet.Packets; @@ -17,30 +18,32 @@ namespace MQTTnet.Server { public sealed class MqttClientConnection : IDisposable { - private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); - private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); - private readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource(); + readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); + readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); + readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource(); - private readonly IMqttRetainedMessagesManager _retainedMessagesManager; - private readonly MqttClientKeepAliveMonitor _keepAliveMonitor; - private readonly MqttClientSessionsManager _sessionsManager; + readonly IMqttRetainedMessagesManager _retainedMessagesManager; + readonly MqttClientKeepAliveMonitor _keepAliveMonitor; + readonly MqttClientSessionsManager _sessionsManager; - private readonly IMqttNetLogger _logger; - private readonly IMqttServerOptions _serverOptions; + readonly IMqttNetLogger _logger; + readonly IMqttServerOptions _serverOptions; - private readonly IMqttChannelAdapter _channelAdapter; - private readonly IMqttDataConverter _dataConverter; - private readonly string _endpoint; - private readonly DateTime _connectedTimestamp; + readonly IMqttChannelAdapter _channelAdapter; + readonly IMqttDataConverter _dataConverter; + readonly string _endpoint; + readonly DateTime _connectedTimestamp; - private Task _packageReceiverTask; - private DateTime _lastPacketReceivedTimestamp; - private DateTime _lastNonKeepAlivePacketReceivedTimestamp; + Task _packageReceiverTask; + DateTime _lastPacketReceivedTimestamp; + DateTime _lastNonKeepAlivePacketReceivedTimestamp; - private long _receivedPacketsCount; - private long _sentPacketsCount = 1; // Start with 1 because the CONNECT packet is not counted anywhere. - private long _receivedApplicationMessagesCount; - private long _sentApplicationMessagesCount; + long _receivedPacketsCount; + long _sentPacketsCount = 1; // Start with 1 because the CONNECT packet is not counted anywhere. + long _receivedApplicationMessagesCount; + long _sentApplicationMessagesCount; + + bool _isTakeover; public MqttClientConnection( MqttConnectPacket connectPacket, @@ -64,7 +67,7 @@ namespace MQTTnet.Server if (logger == null) throw new ArgumentNullException(nameof(logger)); _logger = logger.CreateChildLogger(nameof(MqttClientConnection)); - _keepAliveMonitor = new MqttClientKeepAliveMonitor(ConnectPacket.ClientId, StopAsync, _logger); + _keepAliveMonitor = new MqttClientKeepAliveMonitor(ConnectPacket.ClientId, () => StopAsync(), _logger); _connectedTimestamp = DateTime.UtcNow; _lastPacketReceivedTimestamp = _connectedTimestamp; @@ -77,15 +80,21 @@ namespace MQTTnet.Server public MqttClientSession Session { get; } - public async Task StopAsync() + public bool IsFinalized { get; set; } + + public Task StopAsync(bool isTakeover = false) { + _isTakeover = isTakeover; + StopInternal(); var task = _packageReceiverTask; if (task != null) { - await task.ConfigureAwait(false); + return task; } + + return PlatformAbstractionLayer.CompletedTask; } public void ResetStatistics() @@ -243,11 +252,16 @@ namespace MQTTnet.Server _channelAdapter.ReadingPacketStartedCallback = null; _channelAdapter.ReadingPacketCompletedCallback = null; - _logger.Info("Client '{0}': Session stopped.", ClientId); + _logger.Info("Client '{0}': Connection stopped.", ClientId); _packageReceiverTask = null; } + if (_isTakeover) + { + return MqttClientDisconnectType.Takeover; + } + return disconnectType; } @@ -319,7 +333,7 @@ namespace MQTTnet.Server _sessionsManager.DispatchApplicationMessage(applicationMessage, this); - return Task.FromResult(0); + return PlatformAbstractionLayer.CompletedTask; } Task HandleIncomingPublishPacketWithQoS1Async(MqttPublishPacket publishPacket) @@ -422,9 +436,6 @@ namespace MQTTnet.Server } _logger.Verbose("Queued application message sent (ClientId: {0}).", ClientId); - - // TODO: - //Interlocked.Increment(ref _sentPacketsCount); } } catch (Exception exception) diff --git a/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs b/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs index 9d17552c..6b288251 100644 --- a/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs @@ -9,13 +9,13 @@ namespace MQTTnet.Server { public class MqttClientKeepAliveMonitor { - private readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch(); + readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch(); - private readonly string _clientId; - private readonly Func _keepAliveElapsedCallback; - private readonly IMqttNetLogger _logger; + readonly string _clientId; + readonly Func _keepAliveElapsedCallback; + readonly IMqttNetLogger _logger; - private bool _isPaused; + bool _isPaused; public MqttClientKeepAliveMonitor(string clientId, Func keepAliveElapsedCallback, IMqttNetLogger logger) { @@ -51,7 +51,7 @@ namespace MQTTnet.Server _lastPacketReceivedTracker.Restart(); } - private async Task RunAsync(int keepAlivePeriod, CancellationToken cancellationToken) + async Task RunAsync(int keepAlivePeriod, CancellationToken cancellationToken) { try { diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 08568d81..1e2bcef8 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -16,19 +16,19 @@ namespace MQTTnet.Server { public class MqttClientSessionsManager : Disposable { - private readonly AsyncQueue _messageQueue = new AsyncQueue(); + readonly AsyncQueue _messageQueue = new AsyncQueue(); - private readonly SemaphoreSlim _createConnectionGate = new SemaphoreSlim(1, 1); - private readonly ConcurrentDictionary _connections = new ConcurrentDictionary(); - private readonly ConcurrentDictionary _sessions = new ConcurrentDictionary(); - private readonly IDictionary _serverSessionItems = new ConcurrentDictionary(); + readonly AsyncLock _createConnectionGate = new AsyncLock(); + readonly ConcurrentDictionary _connections = new ConcurrentDictionary(); + readonly ConcurrentDictionary _sessions = new ConcurrentDictionary(); + readonly IDictionary _serverSessionItems = new ConcurrentDictionary(); - private readonly CancellationToken _cancellationToken; - private readonly MqttServerEventDispatcher _eventDispatcher; + readonly CancellationToken _cancellationToken; + readonly MqttServerEventDispatcher _eventDispatcher; - private readonly IMqttRetainedMessagesManager _retainedMessagesManager; - private readonly IMqttServerOptions _options; - private readonly IMqttNetLogger _logger; + readonly IMqttRetainedMessagesManager _retainedMessagesManager; + readonly IMqttServerOptions _options; + readonly IMqttNetLogger _logger; public MqttClientSessionsManager( IMqttServerOptions options, @@ -60,9 +60,11 @@ namespace MQTTnet.Server } } - public Task HandleClientAsync(IMqttChannelAdapter clientAdapter) + public Task HandleClientConnectionAsync(IMqttChannelAdapter clientAdapter) { - return HandleClientAsync(clientAdapter, _cancellationToken); + if (clientAdapter is null) throw new ArgumentNullException(nameof(clientAdapter)); + + return HandleClientConnectionAsync(clientAdapter, _cancellationToken); } public Task> GetClientStatusAsync() @@ -155,7 +157,7 @@ namespace MQTTnet.Server base.Dispose(disposing); } - private async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken) + async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) { @@ -173,7 +175,7 @@ namespace MQTTnet.Server } } - private async Task TryProcessNextQueuedApplicationMessageAsync(CancellationToken cancellationToken) + async Task TryProcessNextQueuedApplicationMessageAsync(CancellationToken cancellationToken) { try { @@ -231,14 +233,14 @@ namespace MQTTnet.Server } } - private async Task HandleClientAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken) + async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken) { var disconnectType = MqttClientDisconnectType.NotClean; string clientId = null; - var clientWasConnected = true; + var clientWasAuthorized = false; MqttConnectPacket connectPacket; - + MqttClientConnection clientConnection = null; try { try @@ -261,7 +263,6 @@ namespace MQTTnet.Server if (connectionValidatorContext.ReasonCode != MqttConnectReasonCode.Success) { - clientWasConnected = false; // Send failure response here without preparing a session. The result for a successful connect // will be sent from the session itself. var connAckPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext); @@ -270,12 +271,13 @@ namespace MQTTnet.Server return; } + clientWasAuthorized = true; clientId = connectPacket.ClientId; - var connection = await CreateConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false); + clientConnection = await CreateClientConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false); await _eventDispatcher.SafeNotifyClientConnectedAsync(clientId).ConfigureAwait(false); - disconnectType = await connection.RunAsync(connectionValidatorContext).ConfigureAwait(false); + disconnectType = await clientConnection.RunAsync(connectionValidatorContext).ConfigureAwait(false); } catch (OperationCanceledException) { @@ -286,8 +288,10 @@ namespace MQTTnet.Server } finally { - if (clientWasConnected) + if (clientWasAuthorized && disconnectType != MqttClientDisconnectType.Takeover) { + // Only cleanup if the client was authorized. If not it will remove the existing connection, session etc. + // This allows to kill connections and sessions from known client IDs. if (clientId != null) { _connections.TryRemove(clientId, out _); @@ -297,18 +301,23 @@ namespace MQTTnet.Server await DeleteSessionAsync(clientId).ConfigureAwait(false); } } + } - await TryCleanupChannelAsync(channelAdapter).ConfigureAwait(false); + await SafeCleanupChannelAsync(channelAdapter).ConfigureAwait(false); - if (clientId != null) - { - await _eventDispatcher.SafeNotifyClientDisconnectedAsync(clientId, disconnectType).ConfigureAwait(false); - } + if (clientWasAuthorized && clientId != null) + { + await _eventDispatcher.SafeNotifyClientDisconnectedAsync(clientId, disconnectType).ConfigureAwait(false); + } + + if (clientConnection != null) + { + clientConnection.IsFinalized = true; } } } - private async Task ValidateConnectionAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter) + async Task ValidateConnectionAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter) { var context = new MqttConnectionValidatorContext(connectPacket, channelAdapter, new ConcurrentDictionary()); @@ -336,17 +345,22 @@ namespace MQTTnet.Server return context; } - private async Task CreateConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter) + async Task CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter) { - await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false); - try + using (await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false)) { var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var session); var isConnectionPresent = _connections.TryGetValue(connectPacket.ClientId, out var existingConnection); if (isConnectionPresent) { - await existingConnection.StopAsync().ConfigureAwait(false); + await existingConnection.StopAsync(true); + + // TODO: This fixes a race condition with unit test Same_Client_Id_Connect_Disconnect_Event_Order. + // It is not clear where the issue is coming from. The connected event is fired BEFORE the disconnected + // event. This is wrong. It seems that the finally block in HandleClientAsync must be finished before we + // can continue here. Maybe there is a better way to do this. + SpinWait.SpinUntil(() => existingConnection.IsFinalized, TimeSpan.FromSeconds(10)); } if (isSessionPresent) @@ -376,13 +390,9 @@ namespace MQTTnet.Server return connection; } - finally - { - _createConnectionGate.Release(); - } } - private async Task InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage) + async Task InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage) { var interceptor = _options.ApplicationMessageInterceptor; if (interceptor == null) @@ -410,7 +420,7 @@ namespace MQTTnet.Server return interceptorContext; } - private async Task TryCleanupChannelAsync(IMqttChannelAdapter channelAdapter) + async Task SafeCleanupChannelAsync(IMqttChannelAdapter channelAdapter) { try { diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 0fa6b4c9..ea687eb4 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -1,15 +1,15 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; -using MQTTnet.Adapter; +using MQTTnet.Adapter; using MQTTnet.Client.Publishing; using MQTTnet.Client.Receiving; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Protocol; using MQTTnet.Server.Status; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Server { @@ -192,7 +192,7 @@ namespace MQTTnet.Server private Task OnHandleClient(IMqttChannelAdapter channelAdapter) { - return _clientSessionsManager.HandleClientAsync(channelAdapter); + return _clientSessionsManager.HandleClientConnectionAsync(channelAdapter); } } }