diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 34e563bc..10da9e58 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -11,11 +11,11 @@ false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). * [Core] Performance optimizations. -* [Core] Fixed a bug which prevents receiving large packets (UWP only) -* [Client] The ManagedClient options now allow configuring the interval for connection checks. -* [Server] Added the Endpoint of the Adapter (remote IP and port) to the connection validation callback. -* [Server] The ipv4 and ipv6 endpoint can be disabled now by setting the bound IP address to _None_. -* [Server] Fix a bug in the keep alive monitor which caused high CPU load (thanks to @GarageGadget). +* [Client] Added support for proxies when using web socket connections (thanks to PitySOFT). +* [Client] Refactored TLS parameter usage and added more parameters (thanks to PitySOFT). +* [Server] Fixed an issue in client keep alive checks (thanks to @jenscski). +* [Server] Changed the order of _ClientConnected_ and _ClientDisconnected_ events so that _ClientConnected_ is fired at first (thanks to @jenscski). +* [Server] Fixed an iussue which lets the server stop processing messages when using the application message interceptor (thanks to @alamsor). Copyright Christian Kratky 2016-2018 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 @@ -56,6 +56,9 @@ + + + \ No newline at end of file diff --git a/README.md b/README.md index 7181d724..1a34eeed 100644 --- a/README.md +++ b/README.md @@ -83,9 +83,10 @@ This project also listed at Open Collective (https://opencollective.com/mqttnet) This library is used in the following projects: +* HA4IoT (Open Source Home Automation system for .NET, ) * MQTT Client Rx (Wrapper for Reactive Extensions, ) * MQTT Tester (MQTT client test app for [Android](https://play.google.com/store/apps/details?id=com.liveowl.mqtttester) and [iOS](https://itunes.apple.com/us/app/mqtt-tester/id1278621826?mt=8)) -* HA4IoT (Open Source Home Automation system for .NET, ) +* Wirehome.Core (Open Source Home Automation system for .NET Core, ) If you use this library and want to see your project here please let me know. diff --git a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs index 985cda7c..85e2b4d8 100644 --- a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs @@ -7,10 +7,11 @@ namespace MQTTnet.Client public class MqttClientOptionsBuilder { private readonly MqttClientOptions _options = new MqttClientOptions(); + private MqttClientTcpOptions _tcpOptions; private MqttClientWebSocketOptions _webSocketOptions; - - private MqttClientTlsOptions _tlsOptions; + private MqttClientOptionsBuilderTlsParameters _tlsParameters; + private MqttClientWebSocketProxyOptions _proxyOptions; public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value) { @@ -76,6 +77,21 @@ namespace MQTTnet.Client return this; } + public MqttClientOptionsBuilder WithProxy(string address, string username = null, string password = null, string domain = null, bool bypassOnLocal = false, string[] bypassList = null) + { + _proxyOptions = new MqttClientWebSocketProxyOptions + { + Address = address, + Username = username, + Password = password, + Domain = domain, + BypassOnLocal = bypassOnLocal, + BypassList = bypassList + }; + + return this; + } + public MqttClientOptionsBuilder WithWebSocketServer(string uri) { _webSocketOptions = new MqttClientWebSocketOptions @@ -86,13 +102,25 @@ namespace MQTTnet.Client return this; } + public MqttClientOptionsBuilder WithTls(MqttClientOptionsBuilderTlsParameters parameters) + { + _tlsParameters = parameters ?? throw new ArgumentNullException(nameof(parameters)); + return this; + } + + public MqttClientOptionsBuilder WithTls() + { + return WithTls(new MqttClientOptionsBuilderTlsParameters { UseTls = true }); + } + + [Obsolete("Use method _WithTls_ which accepts the _MqttClientOptionsBuilderTlsParameters_.")] public MqttClientOptionsBuilder WithTls( bool allowUntrustedCertificates = false, bool ignoreCertificateChainErrors = false, bool ignoreCertificateRevocationErrors = false, params byte[][] certificates) { - _tlsOptions = new MqttClientTlsOptions + _tlsParameters = new MqttClientOptionsBuilderTlsParameters { UseTls = true, AllowUntrustedCertificates = allowUntrustedCertificates, @@ -104,33 +132,47 @@ namespace MQTTnet.Client return this; } - public MqttClientOptionsBuilder WithTls() - { - _tlsOptions = new MqttClientTlsOptions - { - UseTls = true - }; - - return this; - } - public IMqttClientOptions Build() { - if (_tlsOptions != null) + if (_tcpOptions == null && _webSocketOptions == null) { - if (_tcpOptions == null && _webSocketOptions == null) + throw new InvalidOperationException("A channel must be set."); + } + + if (_tlsParameters != null) + { + if (_tlsParameters?.UseTls == true) { - throw new InvalidOperationException("A channel (TCP or WebSocket) must be set if TLS is configured."); + var tlsOptions = new MqttClientTlsOptions + { + UseTls = true, + SslProtocol = _tlsParameters.SslProtocol, + AllowUntrustedCertificates = _tlsParameters.AllowUntrustedCertificates, + Certificates = _tlsParameters.Certificates?.Select(c => c.ToArray()).ToList(), + CertificateValidationCallback = _tlsParameters.CertificateValidationCallback, + IgnoreCertificateChainErrors = _tlsParameters.IgnoreCertificateChainErrors, + IgnoreCertificateRevocationErrors = _tlsParameters.IgnoreCertificateRevocationErrors + }; + + if (_tcpOptions != null) + { + _tcpOptions.TlsOptions = tlsOptions; + } + else if (_webSocketOptions != null) + { + _webSocketOptions.TlsOptions = tlsOptions; + } + } + } + + if (_proxyOptions != null) + { + if (_webSocketOptions == null) + { + throw new InvalidOperationException("Proxies are only supported for WebSocket connections."); } - if (_tcpOptions != null) - { - _tcpOptions.TlsOptions = _tlsOptions; - } - else if (_webSocketOptions != null) - { - _webSocketOptions.TlsOptions = _tlsOptions; - } + _webSocketOptions.ProxyOptions = _proxyOptions; } _options.ChannelOptions = (IMqttClientChannelOptions)_tcpOptions ?? _webSocketOptions; diff --git a/Source/MQTTnet/Client/MqttClientOptionsBuilderTlsParameters.cs b/Source/MQTTnet/Client/MqttClientOptionsBuilderTlsParameters.cs new file mode 100644 index 00000000..5764c797 --- /dev/null +++ b/Source/MQTTnet/Client/MqttClientOptionsBuilderTlsParameters.cs @@ -0,0 +1,29 @@ +using System; +using System.Collections.Generic; +using System.Net.Security; +using System.Security.Authentication; +using System.Security.Cryptography.X509Certificates; + +namespace MQTTnet.Client +{ + public class MqttClientOptionsBuilderTlsParameters + { + public bool UseTls { get; set; } + + public Func CertificateValidationCallback + { + get; + set; + } + + public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12; + + public IEnumerable> Certificates { get; set; } + + public bool AllowUntrustedCertificates { get; set; } + + public bool IgnoreCertificateChainErrors { get; set; } + + public bool IgnoreCertificateRevocationErrors { get; set; } + } +} diff --git a/Source/MQTTnet/Client/MqttClientTlsOptions.cs b/Source/MQTTnet/Client/MqttClientTlsOptions.cs index 2536b231..73d65cbc 100644 --- a/Source/MQTTnet/Client/MqttClientTlsOptions.cs +++ b/Source/MQTTnet/Client/MqttClientTlsOptions.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.Net.Security; using System.Security.Cryptography.X509Certificates; +using System.Security.Authentication; namespace MQTTnet.Client { @@ -17,6 +18,8 @@ namespace MQTTnet.Client public List Certificates { get; set; } + public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12; + public Func CertificateValidationCallback { get; set; } } } diff --git a/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs b/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs index a4dd0d5c..fe8f1d27 100644 --- a/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs +++ b/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs @@ -13,6 +13,8 @@ namespace MQTTnet.Client public CookieContainer CookieContainer { get; set; } + public MqttClientWebSocketProxyOptions ProxyOptions { get; set; } + public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); public override string ToString() diff --git a/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs b/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs new file mode 100644 index 00000000..13b13667 --- /dev/null +++ b/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs @@ -0,0 +1,17 @@ +namespace MQTTnet.Client +{ + public class MqttClientWebSocketProxyOptions + { + public string Address { get; set; } + + public string Username { get; set; } + + public string Password { get; set; } + + public string Domain { get; set; } + + public bool BypassOnLocal { get; set; } + + public string[] BypassList { get; set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index 07451393..bc10d65c 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -63,7 +63,7 @@ namespace MQTTnet.Implementations if (_options.TlsOptions.UseTls) { sslStream = new SslStream(new NetworkStream(_socket, true), false, InternalUserCertificateValidationCallback); - await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), SslProtocols.Tls12, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false); + await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false); } CreateStream(sslStream); diff --git a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs index 3243cd76..53173d27 100644 --- a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs +++ b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs @@ -1,4 +1,5 @@ using System; +using System.Net; using System.Net.WebSockets; using System.Security.Cryptography.X509Certificates; using System.Threading; @@ -14,7 +15,7 @@ namespace MQTTnet.Implementations private readonly MqttClientWebSocketOptions _options; private WebSocket _webSocket; - + public MqttWebSocketChannel(MqttClientWebSocketOptions options) { _options = options ?? throw new ArgumentNullException(nameof(options)); @@ -45,6 +46,11 @@ namespace MQTTnet.Implementations var clientWebSocket = new ClientWebSocket(); + if (_options.ProxyOptions != null) + { + clientWebSocket.Options.Proxy = CreateProxy(); + } + if (_options.RequestHeaders != null) { foreach (var requestHeader in _options.RequestHeaders) @@ -132,5 +138,31 @@ namespace MQTTnet.Implementations _webSocket = null; } } + + private IWebProxy CreateProxy() + { + if (string.IsNullOrEmpty(_options.ProxyOptions?.Address)) + { + return null; + } + +#if WINDOWS_UWP + throw new NotSupportedException("Proxies are not supported in UWP."); +#elif NETSTANDARD1_3 + throw new NotSupportedException("Proxies are not supported in netstandard 1.3."); +#else + var proxyUri = new Uri(_options.ProxyOptions.Address); + + if (!string.IsNullOrEmpty(_options.ProxyOptions.Username) && !string.IsNullOrEmpty(_options.ProxyOptions.Password)) + { + var credentials = + new NetworkCredential(_options.ProxyOptions.Username, _options.ProxyOptions.Password, _options.ProxyOptions.Domain); + + return new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList, credentials); + } + + return new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList); +#endif + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/IMqttClientSession.cs b/Source/MQTTnet/Server/IMqttClientSession.cs index a94ad185..9fdb0eb9 100644 --- a/Source/MQTTnet/Server/IMqttClientSession.cs +++ b/Source/MQTTnet/Server/IMqttClientSession.cs @@ -14,7 +14,7 @@ namespace MQTTnet.Server void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket); void ClearPendingApplicationMessages(); - Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter); + Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter); void Stop(MqttClientDisconnectType disconnectType); Task SubscribeAsync(IList topicFilters); diff --git a/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs b/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs index 8afb7acd..f1a209b4 100644 --- a/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs @@ -14,9 +14,9 @@ namespace MQTTnet.Server private readonly IMqttClientSession _clientSession; private readonly IMqttNetChildLogger _logger; - + private bool _isPaused; - + public MqttClientKeepAliveMonitor(IMqttClientSession clientSession, IMqttNetChildLogger logger) { if (logger == null) throw new ArgumentNullException(nameof(logger)); @@ -76,7 +76,7 @@ namespace MQTTnet.Server { _logger.Warning(null, "Client '{0}': Did not receive any packet or keep alive signal.", _clientSession.ClientId); _clientSession.Stop(MqttClientDisconnectType.NotClean); - + return; } @@ -96,7 +96,7 @@ namespace MQTTnet.Server } finally { - _logger.Verbose("Client {0}: Stopped checking keep alive timeout.", _clientSession.ClientId); + _logger.Verbose("Client '{0}': Stopped checking keep alive timeout.", _clientSession.ClientId); } } } diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 8ddb8455..ffc9fbdb 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -67,13 +67,13 @@ namespace MQTTnet.Server status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived; } - public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) + public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) { _workerTask = RunInternalAsync(connectPacket, adapter); return _workerTask; } - private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) + private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) { if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket)); if (adapter == null) throw new ArgumentNullException(nameof(adapter)); @@ -138,8 +138,6 @@ namespace MQTTnet.Server _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; } - - return _wasCleanDisconnect; } private async Task Cleanup() @@ -199,6 +197,8 @@ namespace MQTTnet.Server finally { _logger.Info("Client '{0}': Session stopped.", ClientId); + + _sessionsManager.Server.OnClientDisconnected(ClientId, _wasCleanDisconnect); } } @@ -447,12 +447,12 @@ namespace MQTTnet.Server private void OnAdapterReadingPacketCompleted(object sender, EventArgs e) { - _keepAliveMonitor?.Pause(); + _keepAliveMonitor?.Resume(); } private void OnAdapterReadingPacketStarted(object sender, EventArgs e) { - _keepAliveMonitor?.Resume(); + _keepAliveMonitor?.Pause(); } } } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 6b62e44b..d8f4f141 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -44,7 +44,7 @@ namespace MQTTnet.Server public void Start() { - Task.Factory.StartNew(() => ProcessQueuedApplicationMessages(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default); + Task.Factory.StartNew(() => TryProcessQueuedApplicationMessages(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default); } public void Stop() @@ -125,6 +125,7 @@ namespace MQTTnet.Server { _sessions.Remove(clientId); } + _logger.Verbose("Session for client '{0}' deleted.", clientId); } @@ -133,54 +134,69 @@ namespace MQTTnet.Server _messageQueue?.Dispose(); } - private void ProcessQueuedApplicationMessages(CancellationToken cancellationToken) + private void TryProcessQueuedApplicationMessages(CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) { try { - var enqueuedApplicationMessage = _messageQueue.Take(cancellationToken); - var sender = enqueuedApplicationMessage.Sender; - var applicationMessage = enqueuedApplicationMessage.PublishPacket.ToApplicationMessage(); - - var interceptorContext = InterceptApplicationMessage(sender, applicationMessage); - if (interceptorContext != null) - { - if (interceptorContext.CloseConnection) - { - enqueuedApplicationMessage.Sender.Stop(MqttClientDisconnectType.NotClean); - } - - if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish) - { - return; - } - - applicationMessage = interceptorContext.ApplicationMessage; - } - - Server.OnApplicationMessageReceived(sender?.ClientId, applicationMessage); - - if (applicationMessage.Retain) - { - _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).GetAwaiter().GetResult(); - } - - foreach (var clientSession in GetSessions()) - { - clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage.ToPublishPacket()); - } + TryProcessNextQueuedApplicationMessage(cancellationToken); } catch (OperationCanceledException) { } catch (Exception exception) { - _logger.Error(exception, "Unhandled exception while processing queued application message."); + _logger.Error(exception, "Unhandled exception while processing queued application messages."); } } } + private void TryProcessNextQueuedApplicationMessage(CancellationToken cancellationToken) + { + try + { + var enqueuedApplicationMessage = _messageQueue.Take(cancellationToken); + var sender = enqueuedApplicationMessage.Sender; + var applicationMessage = enqueuedApplicationMessage.PublishPacket.ToApplicationMessage(); + + var interceptorContext = InterceptApplicationMessage(sender, applicationMessage); + if (interceptorContext != null) + { + if (interceptorContext.CloseConnection) + { + enqueuedApplicationMessage.Sender.Stop(MqttClientDisconnectType.NotClean); + } + + if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish) + { + return; + } + + applicationMessage = interceptorContext.ApplicationMessage; + } + + Server.OnApplicationMessageReceived(sender?.ClientId, applicationMessage); + + if (applicationMessage.Retain) + { + _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).GetAwaiter().GetResult(); + } + + foreach (var clientSession in GetSessions()) + { + clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage.ToPublishPacket()); + } + } + catch (OperationCanceledException) + { + } + catch (Exception exception) + { + _logger.Error(exception, "Unhandled exception while processing next queued application message."); + } + } + private List GetSessions() { lock (_sessions) @@ -192,8 +208,7 @@ namespace MQTTnet.Server private async Task RunSession(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken) { var clientId = string.Empty; - var wasCleanDisconnect = false; - + try { var firstPacket = await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false); @@ -238,7 +253,7 @@ namespace MQTTnet.Server Server.OnClientConnected(clientId); - wasCleanDisconnect = await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false); + await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false); } catch (OperationCanceledException) { @@ -253,8 +268,6 @@ namespace MQTTnet.Server { DeleteSession(clientId); } - - Server.OnClientDisconnected(clientId, wasCleanDisconnect); } } diff --git a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs index 2f1c5e99..be899c31 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Net.Sockets; using System.Threading; using System.Threading.Tasks; @@ -8,13 +8,14 @@ using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Packets; +using MQTTnet.Implementations; +using MQTTnet.Server; namespace MQTTnet.Core.Tests { [TestClass] public class MqttClientTests { - [TestMethod] public async Task ClientDisconnectException() { diff --git a/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs b/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs index 267b7bca..b18904a0 100644 --- a/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs @@ -60,7 +60,7 @@ namespace MQTTnet.Core.Tests { public string ClientId { get; } - public int StopCalledCount { get; set; } + public int StopCalledCount { get; private set; } public void FillStatus(MqttClientSessionStatus status) { @@ -77,7 +77,7 @@ namespace MQTTnet.Core.Tests throw new NotSupportedException(); } - public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) + public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) { throw new NotSupportedException(); } diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 03888668..0c34bdeb 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -552,6 +552,54 @@ namespace MQTTnet.Core.Tests Assert.IsTrue(bodyIsMatching); } + [TestMethod] + public async Task MqttServer_SameClientIdConnectDisconnectEventOrder() + { + var serverAdapter = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger()); + var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger()); + + var connectedClient = false; + var connecteCalledBeforeConnectedClients = false; + + s.ClientConnected += (_, __) => + { + connecteCalledBeforeConnectedClients |= connectedClient; + connectedClient = true; + }; + + s.ClientDisconnected += (_, __) => + { + connectedClient = false; + }; + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost") + .WithClientId(Guid.NewGuid().ToString()) + .Build(); + + await s.StartAsync(new MqttServerOptions()); + + var c1 = new MqttFactory().CreateMqttClient(); + var c2 = new MqttFactory().CreateMqttClient(); + + await c1.ConnectAsync(clientOptions); + + await Task.Delay(100); + + await c2.ConnectAsync(clientOptions); + + await Task.Delay(100); + + await c1.DisconnectAsync(); + await c2.DisconnectAsync(); + + await s.StopAsync(); + + await Task.Delay(100); + + Assert.IsFalse(connecteCalledBeforeConnectedClients, "ClientConnected was called before ClientDisconnect was called"); + } + private class TestStorage : IMqttServerStorage { public IList Messages = new List();