From b09eb1213c699d00d15cd0fa7f8e3e03672a74f9 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sun, 8 Jul 2018 22:04:06 +0200 Subject: [PATCH 01/31] Refactor logging. --- Source/MQTTnet/Implementations/MqttTcpServerListener.cs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs index 7f9ffbd3..e3522979 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs @@ -78,8 +78,10 @@ namespace MQTTnet.Implementations await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false); } - var protocol = _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6"; - _logger.Verbose($"Client '{clientSocket.RemoteEndPoint}' accepted by TCP listener '{_socket.LocalEndPoint}, {protocol}'."); + _logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.", + clientSocket.RemoteEndPoint, + _socket.LocalEndPoint, + _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6"); var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _logger); ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); From fa34908d82a56df572d1a7d372927bb7f8d58b22 Mon Sep 17 00:00:00 2001 From: Marius Ruginosu Date: Wed, 18 Jul 2018 10:51:45 +0300 Subject: [PATCH 02/31] Add proxy settings for WebSocket connections for .NET 451 and 461 --- .../Client/MqttClientOptionsBuilder.cs | 23 ++++++++++++++++ .../Client/MqttClientWebSocketOptions.cs | 4 +++ .../Client/MqttClientWebSocketProxyOptions.cs | 21 +++++++++++++++ .../Implementations/MqttWebSocketChannel.cs | 26 ++++++++++++++++++- 4 files changed, 73 insertions(+), 1 deletion(-) create mode 100644 Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs diff --git a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs index 985cda7c..4948ad37 100644 --- a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs @@ -76,6 +76,29 @@ namespace MQTTnet.Client return this; } +#if NET452 || NET461 + + public MqttClientOptionsBuilder WithProxy(string address, string username = null, string password = null, string domain = null, bool bypassOnLocal = false, string[] bypassList = null) + { + if (_webSocketOptions == null) + { + throw new InvalidOperationException("A WebSocket channel must be set if MqttClientWebSocketProxy is configured."); + } + + _webSocketOptions.MqttClientWebSocketProxy = new MqttClientWebSocketProxyOptions + { + Address = address, + Username = username, + Password = password, + Domain = domain, + BypassOnLocal = bypassOnLocal, + BypassList = bypassList + }; + + return this; + } +#endif + public MqttClientOptionsBuilder WithWebSocketServer(string uri) { _webSocketOptions = new MqttClientWebSocketOptions diff --git a/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs b/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs index a4dd0d5c..2cce4310 100644 --- a/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs +++ b/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs @@ -13,6 +13,10 @@ namespace MQTTnet.Client public CookieContainer CookieContainer { get; set; } +#if NET452 || NET461 + public MqttClientWebSocketProxyOptions MqttClientWebSocketProxy { get; set; } = new MqttClientWebSocketProxyOptions(); +#endif + 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..81f48da8 --- /dev/null +++ b/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs @@ -0,0 +1,21 @@ +#if NET452 || NET461 + +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; } = true; + + public string[] BypassList { get; set; } + } +} + +#endif \ No newline at end of file diff --git a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs index 3243cd76..cf8c183a 100644 --- a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs +++ b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs @@ -6,6 +6,10 @@ using System.Threading.Tasks; using MQTTnet.Channel; using MQTTnet.Client; +#if NET452 || NET461 + using System.Net; +#endif + namespace MQTTnet.Implementations { public class MqttWebSocketChannel : IMqttChannel @@ -14,7 +18,7 @@ namespace MQTTnet.Implementations private readonly MqttClientWebSocketOptions _options; private WebSocket _webSocket; - + public MqttWebSocketChannel(MqttClientWebSocketOptions options) { _options = options ?? throw new ArgumentNullException(nameof(options)); @@ -45,6 +49,26 @@ namespace MQTTnet.Implementations var clientWebSocket = new ClientWebSocket(); +#if NET452 || NET461 + // use proxy if we have an address + if (string.IsNullOrEmpty(_options.MqttClientWebSocketProxy.Address) == false) + { + var proxyUri = new Uri(_options.MqttClientWebSocketProxy.Address); + + // use proxy credentials if we have them + if (string.IsNullOrEmpty(_options.MqttClientWebSocketProxy.Username) == false && string.IsNullOrEmpty(_options.MqttClientWebSocketProxy.Password) == false) + { + var credentials = new NetworkCredential(_options.MqttClientWebSocketProxy.Username, _options.MqttClientWebSocketProxy.Password, _options.MqttClientWebSocketProxy.Domain); + clientWebSocket.Options.Proxy = new WebProxy(proxyUri, _options.MqttClientWebSocketProxy.BypassOnLocal, _options.MqttClientWebSocketProxy.BypassList, credentials); + } + else + { + // use proxy without credentials + clientWebSocket.Options.Proxy = new WebProxy(proxyUri, _options.MqttClientWebSocketProxy.BypassOnLocal, _options.MqttClientWebSocketProxy.BypassList); + } + } +#endif + if (_options.RequestHeaders != null) { foreach (var requestHeader in _options.RequestHeaders) From 65a8b091b9f840bb0bb5edbfe00a518d718887aa Mon Sep 17 00:00:00 2001 From: Marius Ruginosu Date: Wed, 18 Jul 2018 12:19:09 +0300 Subject: [PATCH 03/31] Add option to specify the SSLProtocol and CertificateValidationCallback in TLS options, SSLProtocol default is TLS1.2 --- Source/MQTTnet/Client/MqttClientOptionsBuilder.cs | 11 +++++++++-- Source/MQTTnet/Client/MqttClientTlsOptions.cs | 3 +++ Source/MQTTnet/Implementations/MqttTcpChannel.cs | 2 +- 3 files changed, 13 insertions(+), 3 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs index 4948ad37..77c61bd3 100644 --- a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs @@ -1,5 +1,8 @@ using System; using System.Linq; +using System.Net.Security; +using System.Security.Authentication; +using System.Security.Cryptography.X509Certificates; using MQTTnet.Serializer; namespace MQTTnet.Client @@ -113,7 +116,9 @@ namespace MQTTnet.Client bool allowUntrustedCertificates = false, bool ignoreCertificateChainErrors = false, bool ignoreCertificateRevocationErrors = false, - params byte[][] certificates) + byte[][] certificates = null, + SslProtocols sslProtocol = SslProtocols.Tls12, + Func certificateValidationCallback = null) { _tlsOptions = new MqttClientTlsOptions { @@ -121,7 +126,9 @@ namespace MQTTnet.Client AllowUntrustedCertificates = allowUntrustedCertificates, IgnoreCertificateChainErrors = ignoreCertificateChainErrors, IgnoreCertificateRevocationErrors = ignoreCertificateRevocationErrors, - Certificates = certificates?.ToList() + Certificates = certificates?.ToList(), + SslProtocol = sslProtocol, + CertificateValidationCallback = certificateValidationCallback }; return this; 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/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index c48a06f6..d7c55a77 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); From 6e995230a78d1e8efe1432dc1373fc816e149317 Mon Sep 17 00:00:00 2001 From: Marius Ruginosu Date: Wed, 18 Jul 2018 16:50:29 +0300 Subject: [PATCH 04/31] Created overload methods for WithTls default values --- .../Client/MqttClientOptionsBuilder.cs | 52 +++++++++++++++---- 1 file changed, 42 insertions(+), 10 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs index 77c61bd3..ed8ae03d 100644 --- a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs @@ -112,8 +112,49 @@ namespace MQTTnet.Client return this; } + + public MqttClientOptionsBuilder WithTls() + { + return WithTls(null); + } + + public MqttClientOptionsBuilder WithTls(Func certificateValidationCallback) + { + return WithTls(SslProtocols.None, certificateValidationCallback); + } + + public MqttClientOptionsBuilder WithTls(SslProtocols sslProtocol, + Func certificateValidationCallback = null) + { + return WithTls(new byte[][] { }, sslProtocol, certificateValidationCallback); + } + + public MqttClientOptionsBuilder WithTls(byte[][] certificates, + SslProtocols sslProtocol = SslProtocols.Tls12, + Func certificateValidationCallback = null) + { + return WithTls(false, certificates, sslProtocol, certificateValidationCallback); + } + + public MqttClientOptionsBuilder WithTls(bool ignoreCertificateRevocationErrors, + byte[][] certificates = null, + SslProtocols sslProtocol = SslProtocols.Tls12, + Func certificateValidationCallback = null) + { + return WithTls(false, ignoreCertificateRevocationErrors, certificates, sslProtocol, certificateValidationCallback); + } + + public MqttClientOptionsBuilder WithTls(bool ignoreCertificateChainErrors, + bool ignoreCertificateRevocationErrors = false, + byte[][] certificates = null, + SslProtocols sslProtocol = SslProtocols.Tls12, + Func certificateValidationCallback = null) + { + return WithTls(false, ignoreCertificateChainErrors, ignoreCertificateRevocationErrors, certificates, sslProtocol, certificateValidationCallback); + } + public MqttClientOptionsBuilder WithTls( - bool allowUntrustedCertificates = false, + bool allowUntrustedCertificates, bool ignoreCertificateChainErrors = false, bool ignoreCertificateRevocationErrors = false, byte[][] certificates = null, @@ -134,15 +175,6 @@ namespace MQTTnet.Client return this; } - public MqttClientOptionsBuilder WithTls() - { - _tlsOptions = new MqttClientTlsOptions - { - UseTls = true - }; - - return this; - } public IMqttClientOptions Build() { From 740a35b018b44cb5bbe569062e691724b10e6b88 Mon Sep 17 00:00:00 2001 From: Jens-Christian Skibakk Date: Wed, 18 Jul 2018 19:54:07 +0200 Subject: [PATCH 05/31] Skip build uap10.0 when using dotnet cli --- .../MQTTnet.Extensions.ManagedClient.csproj | 3 ++- Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj | 3 ++- Source/MQTTnet/MQTTnet.csproj | 5 +++-- 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj b/Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj index 1a3d9cc9..c8405a2a 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj +++ b/Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj @@ -1,7 +1,8 @@  - netstandard1.3;netstandard2.0;net452;net461;uap10.0 + netstandard1.3;netstandard2.0;net452;net461 + $(TargetFrameworks);uap10.0 diff --git a/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj b/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj index 1a3d9cc9..c8405a2a 100644 --- a/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj +++ b/Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj @@ -1,7 +1,8 @@  - netstandard1.3;netstandard2.0;net452;net461;uap10.0 + netstandard1.3;netstandard2.0;net452;net461 + $(TargetFrameworks);uap10.0 diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index 984b5630..30eace10 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -1,8 +1,9 @@  - netstandard1.3;netstandard2.0;net452;net461;uap10.0 - netstandard1.3;netstandard2.0 + netstandard1.3;netstandard2.0 + $(TargetFrameworks);net452;net461 + $(TargetFrameworks);uap10.0 MQTTnet MQTTnet False From 21048b51ae673d04e91997bd71cf9d56d9697e10 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 1 Jul 2018 19:12:18 +0200 Subject: [PATCH 06/31] prove there is something wrong --- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 26 +++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 95bb67da..14f1076a 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -233,6 +233,32 @@ namespace MQTTnet.Core.Tests await c1.PublishAsync(message); } } + + [TestMethod] + public async Task MqttServer_ShutdownDisconnectsClientsGracefully() + { + var serverAdapter = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger()); + var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger()); + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost") + .Build(); + + bool disconnectCalled = false; + + await s.StartAsync(new MqttServerOptions()); + + var c1 = new MqttFactory().CreateMqttClient(); + c1.Disconnected += (sender, args) => disconnectCalled = true; + + await c1.ConnectAsync(clientOptions); + + await s.StopAsync(); + + await Task.Delay(500); + + Assert.IsTrue(disconnectCalled); + } [TestMethod] public async Task MqttServer_RetainedMessagesFlow() From b55f5ff2988ad5977da90f8d08b64280ba37e174 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 1 Jul 2018 20:27:42 +0200 Subject: [PATCH 07/31] event is triggered but something is blocking --- Source/MQTTnet/Client/MqttClient.cs | 7 ++++-- .../MQTTnet/Implementations/MqttTcpChannel.cs | 23 ++++++++++++------- Source/MQTTnet/Server/MqttClientSession.cs | 21 ++++++++++++++++- .../Server/MqttClientSessionsManager.cs | 10 -------- Source/MQTTnet/Server/MqttServer.cs | 4 ++-- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 4 +++- 6 files changed, 45 insertions(+), 24 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index ccedb5ae..93112598 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -222,8 +222,11 @@ namespace MQTTnet.Client try { - await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false); - await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false); + if (!(exception is MqttCommunicationClosedGracefullyException)) + { + await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false); + await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false); + } if (_adapter != null) { diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index c48a06f6..07451393 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -87,8 +87,14 @@ namespace MQTTnet.Implementations public void Dispose() { - TryDispose(_stream, () => _stream = null); - TryDispose(_socket, () => _socket = null); + Cleanup(ref _stream, (s) => s.Dispose()); + Cleanup(ref _socket, (s) => { + if (s.Connected) + { + s.Shutdown(SocketShutdown.Both); + } + s.Dispose(); + }); } private bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors) @@ -157,11 +163,16 @@ namespace MQTTnet.Implementations } } - private static void TryDispose(IDisposable disposable, Action afterDispose) + private static void Cleanup(ref T item, Action handler) where T : class { + var temp = item; + item = null; try { - disposable?.Dispose(); + if (temp != null) + { + handler(temp); + } } catch (ObjectDisposedException) { @@ -169,10 +180,6 @@ namespace MQTTnet.Implementations catch (NullReferenceException) { } - finally - { - afterDispose(); - } } } } diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 429fdb45..05d552cb 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -29,6 +29,7 @@ namespace MQTTnet.Server private MqttApplicationMessage _willMessage; private bool _wasCleanDisconnect; private IMqttChannelAdapter _adapter; + private Task _run; public MqttClientSession( string clientId, @@ -65,7 +66,13 @@ namespace MQTTnet.Server status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived; } - public async Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) + public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) + { + _run = RunInternalAsync(connectPacket, adapter); + return _run; + } + + private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) { if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket)); if (adapter == null) throw new ArgumentNullException(nameof(adapter)); @@ -122,6 +129,16 @@ namespace MQTTnet.Server { _adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted; _adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted; + + try + { + await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); + _adapter.Dispose(); + } + catch (Exception exception) + { + _logger.Error(exception, exception.Message); + } } _adapter = null; @@ -153,6 +170,8 @@ namespace MQTTnet.Server } _willMessage = null; + + _run?.GetAwaiter().GetResult(); } finally { diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 34080a00..6b62e44b 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -249,16 +249,6 @@ namespace MQTTnet.Server } finally { - try - { - await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); - clientAdapter.Dispose(); - } - catch (Exception exception) - { - _logger.Error(exception, exception.Message); - } - if (!_options.EnablePersistentSessions) { DeleteSession(clientId); diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index dd732f1b..6d532917 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -106,14 +106,14 @@ namespace MQTTnet.Server _cancellationTokenSource.Cancel(false); + _clientSessionsManager.Stop(); + foreach (var adapter in _adapters) { adapter.ClientAccepted -= OnClientAccepted; await adapter.StopAsync().ConfigureAwait(false); } - _clientSessionsManager.Stop(); - _logger.Info("Stopped."); Stopped?.Invoke(this, EventArgs.Empty); } diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 14f1076a..24626ab8 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -253,9 +253,11 @@ namespace MQTTnet.Core.Tests await c1.ConnectAsync(clientOptions); + await Task.Delay(500); + await s.StopAsync(); - await Task.Delay(500); + await Task.Delay(1000); Assert.IsTrue(disconnectCalled); } From 53bd2a0960666ec6d3a28aefddbf3b729f2634f9 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 1 Jul 2018 21:24:48 +0200 Subject: [PATCH 08/31] fixed blocking code in networkstrem and reduced sleep in tests --- Source/MQTTnet/Server/MqttClientSession.cs | 32 ++++++++++++++------- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 4 +-- 2 files changed, 24 insertions(+), 12 deletions(-) diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 05d552cb..204bdea8 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -30,6 +30,7 @@ namespace MQTTnet.Server private bool _wasCleanDisconnect; private IMqttChannelAdapter _adapter; private Task _run; + private IDisposable _cleanupHandle; public MqttClientSession( string clientId, @@ -84,6 +85,10 @@ namespace MQTTnet.Server adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted; _cancellationTokenSource = new CancellationTokenSource(); + + //woraround for https://github.com/dotnet/corefx/issues/24430 + _cleanupHandle = _cancellationTokenSource.Token.Register(() => Cleanup()); + //endworkaround _wasCleanDisconnect = false; _willMessage = connectPacket.WillMessage; @@ -129,20 +134,14 @@ namespace MQTTnet.Server { _adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted; _adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted; - - try - { - await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); - _adapter.Dispose(); - } - catch (Exception exception) - { - _logger.Error(exception, exception.Message); - } + await Cleanup(); } _adapter = null; + _cleanupHandle?.Dispose(); + _cleanupHandle = null; + _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; } @@ -150,6 +149,19 @@ namespace MQTTnet.Server return _wasCleanDisconnect; } + private async Task Cleanup() + { + try + { + await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); + _adapter.Dispose(); + } + catch (Exception exception) + { + _logger.Error(exception, exception.Message); + } + } + public void Stop(MqttClientDisconnectType type) { try diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 24626ab8..39259d0f 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -253,11 +253,11 @@ namespace MQTTnet.Core.Tests await c1.ConnectAsync(clientOptions); - await Task.Delay(500); + await Task.Delay(100); await s.StopAsync(); - await Task.Delay(1000); + await Task.Delay(100); Assert.IsTrue(disconnectCalled); } From ae80de26ec046ca92fc7ea6151f2ed30793c18bb Mon Sep 17 00:00:00 2001 From: JanEggers Date: Thu, 5 Jul 2018 19:41:58 +0200 Subject: [PATCH 09/31] addressed feedback --- Source/MQTTnet/Server/MqttClientSession.cs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 204bdea8..8ef401c2 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -29,7 +29,7 @@ namespace MQTTnet.Server private MqttApplicationMessage _willMessage; private bool _wasCleanDisconnect; private IMqttChannelAdapter _adapter; - private Task _run; + private Task _workerTask; private IDisposable _cleanupHandle; public MqttClientSession( @@ -69,8 +69,8 @@ namespace MQTTnet.Server public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) { - _run = RunInternalAsync(connectPacket, adapter); - return _run; + _workerTask = RunInternalAsync(connectPacket, adapter); + return _workerTask; } private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) @@ -134,7 +134,7 @@ namespace MQTTnet.Server { _adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted; _adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted; - await Cleanup(); + await Cleanup().ConfigureAwait(false); } _adapter = null; @@ -183,7 +183,7 @@ namespace MQTTnet.Server _willMessage = null; - _run?.GetAwaiter().GetResult(); + _workerTask?.GetAwaiter().GetResult(); } finally { From 0dea9e6a2bed220a8d462529810c271c5b5285e7 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 7 Jul 2018 09:11:52 +0200 Subject: [PATCH 10/31] changed waiting condition --- Source/MQTTnet/Client/MqttClient.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 93112598..7bc21de4 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -222,7 +222,7 @@ namespace MQTTnet.Client try { - if (!(exception is MqttCommunicationClosedGracefullyException)) + if (clientWasConnected) { await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false); await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false); From e6cfef5295482002671e34fdeec48913c09c0c18 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Wed, 18 Jul 2018 21:13:02 +0200 Subject: [PATCH 11/31] fixed disconnect to be triggered just once --- Source/MQTTnet/Client/MqttClient.cs | 24 ++++++++++++++++----- Tests/MQTTnet.Core.Tests/MqttClientTests.cs | 13 ++++++----- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 6 +++--- 3 files changed, 30 insertions(+), 13 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 7bc21de4..86264e40 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -29,6 +29,7 @@ namespace MQTTnet.Client private Task _keepAliveMessageSenderTask; private IMqttChannelAdapter _adapter; private bool _cleanDisconnectInitiated; + private TaskCompletionSource _disconnectReason; public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger) { @@ -54,6 +55,7 @@ namespace MQTTnet.Client try { _cancellationTokenSource = new CancellationTokenSource(); + _disconnectReason = new TaskCompletionSource(); _options = options; _packetIdentifierProvider.Reset(); _packetDispatcher.Reset(); @@ -85,8 +87,10 @@ namespace MQTTnet.Client catch (Exception exception) { _logger.Error(exception, "Error while connecting with server."); - await DisconnectInternalAsync(null, exception).ConfigureAwait(false); - + if (_disconnectReason.TrySetException(exception)) + { + await DisconnectInternalAsync(null, exception).ConfigureAwait(false); + } throw; } } @@ -104,7 +108,10 @@ namespace MQTTnet.Client } finally { - await DisconnectInternalAsync(null, null).ConfigureAwait(false); + if (_disconnectReason.TrySetCanceled()) + { + await DisconnectInternalAsync(null, null).ConfigureAwait(false); + } } } @@ -355,7 +362,10 @@ namespace MQTTnet.Client _logger.Error(exception, "Unhandled exception while sending/receiving keep alive packets."); } - await DisconnectInternalAsync(_keepAliveMessageSenderTask, exception).ConfigureAwait(false); + if (_disconnectReason.TrySetException(exception)) + { + await DisconnectInternalAsync(_keepAliveMessageSenderTask, exception).ConfigureAwait(false); + } } finally { @@ -398,7 +408,11 @@ namespace MQTTnet.Client } _packetDispatcher.Dispatch(exception); - await DisconnectInternalAsync(_packetReceiverTask, exception).ConfigureAwait(false); + + if (_disconnectReason.TrySetException(exception)) + { + await DisconnectInternalAsync(_packetReceiverTask, exception).ConfigureAwait(false); + } } finally { diff --git a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs index 1334a803..455cf7dc 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs @@ -1,4 +1,5 @@ -using System.Net.Sockets; +using System; +using System.Net.Sockets; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; @@ -16,10 +17,10 @@ namespace MQTTnet.Core.Tests var factory = new MqttFactory(); var client = factory.CreateMqttClient(); - var exceptionIsCorrect = false; + Exception ex = null; client.Disconnected += (s, e) => { - exceptionIsCorrect = e.Exception is MqttCommunicationException c && c.InnerException is SocketException; + ex = e.Exception; }; try @@ -29,8 +30,10 @@ namespace MQTTnet.Core.Tests catch { } - - Assert.IsTrue(exceptionIsCorrect); + + Assert.IsNotNull(ex); + Assert.IsInstanceOfType(ex, typeof(MqttCommunicationException)); + Assert.IsInstanceOfType(ex.InnerException, typeof(SocketException)); } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 39259d0f..117e6ef9 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -244,12 +244,12 @@ namespace MQTTnet.Core.Tests .WithTcpServer("localhost") .Build(); - bool disconnectCalled = false; + var disconnectCalled = 0; await s.StartAsync(new MqttServerOptions()); var c1 = new MqttFactory().CreateMqttClient(); - c1.Disconnected += (sender, args) => disconnectCalled = true; + c1.Disconnected += (sender, args) => disconnectCalled++; await c1.ConnectAsync(clientOptions); @@ -259,7 +259,7 @@ namespace MQTTnet.Core.Tests await Task.Delay(100); - Assert.IsTrue(disconnectCalled); + Assert.AreEqual(1, disconnectCalled); } [TestMethod] From f7ea2d29e1867602f85c110cb3e3a784de3fc0d6 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Thu, 19 Jul 2018 19:14:36 +0200 Subject: [PATCH 12/31] prove its broken --- Source/MQTTnet/Client/MqttClient.cs | 4 +-- Source/MQTTnet/Properties/AssemblyInfo.cs | 3 ++ Tests/MQTTnet.Core.Tests/MqttClientTests.cs | 36 +++++++++++++++++++++ 3 files changed, 41 insertions(+), 2 deletions(-) create mode 100644 Source/MQTTnet/Properties/AssemblyInfo.cs diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 86264e40..4891b1f0 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -25,8 +25,8 @@ namespace MQTTnet.Client private IMqttClientOptions _options; private CancellationTokenSource _cancellationTokenSource; - private Task _packetReceiverTask; - private Task _keepAliveMessageSenderTask; + internal Task _packetReceiverTask; + internal Task _keepAliveMessageSenderTask; private IMqttChannelAdapter _adapter; private bool _cleanDisconnectInitiated; private TaskCompletionSource _disconnectReason; diff --git a/Source/MQTTnet/Properties/AssemblyInfo.cs b/Source/MQTTnet/Properties/AssemblyInfo.cs new file mode 100644 index 00000000..8259cc15 --- /dev/null +++ b/Source/MQTTnet/Properties/AssemblyInfo.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly:InternalsVisibleTo("MQTTnet.Core.Tests")] diff --git a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs index 455cf7dc..2f1c5e99 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs @@ -1,9 +1,13 @@ using System; using System.Net.Sockets; +using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Adapter; using MQTTnet.Client; +using MQTTnet.Diagnostics; using MQTTnet.Exceptions; +using MQTTnet.Packets; namespace MQTTnet.Core.Tests { @@ -35,5 +39,37 @@ namespace MQTTnet.Core.Tests Assert.IsInstanceOfType(ex, typeof(MqttCommunicationException)); Assert.IsInstanceOfType(ex.InnerException, typeof(SocketException)); } + + [TestMethod] + public async Task ClientCleanupOnAuthentificationFails() + { + var channel = new TestMqttCommunicationAdapter(); + var channel2 = new TestMqttCommunicationAdapter(); + channel.Partner = channel2; + channel2.Partner = channel; + + Task.Run(async () => { + var connect = await channel2.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None); + await channel2.SendPacketAsync(new MqttConnAckPacket() { ConnectReturnCode = Protocol.MqttConnectReturnCode.ConnectionRefusedNotAuthorized }, CancellationToken.None); + }); + + + + var fake = new TestMqttCommunicationAdapterFactory(channel); + + var client = new MqttClient(fake, new MqttNetLogger()); + + try + { + await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("any-server").Build()); + } + catch (Exception ex) + { + Assert.IsInstanceOfType(ex, typeof(MqttConnectingFailedException)); + } + + Assert.IsTrue(client._packetReceiverTask == null || client._packetReceiverTask.IsCompleted, "receive loop not completed"); + Assert.IsTrue(client._keepAliveMessageSenderTask == null || client._keepAliveMessageSenderTask.IsCompleted, "keepalive loop not completed"); + } } } From d30fe612646043909d55ca5f181fbb474a6364a2 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Thu, 19 Jul 2018 19:15:48 +0200 Subject: [PATCH 13/31] fixed missing waits --- Source/MQTTnet/Client/MqttClient.cs | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 4891b1f0..0fa074c2 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -229,11 +229,8 @@ namespace MQTTnet.Client try { - if (clientWasConnected) - { - await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false); - await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false); - } + await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false); + await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false); if (_adapter != null) { From 55e8927ab7d292ee1a6c6dbff4804795ce1419c1 Mon Sep 17 00:00:00 2001 From: Jens-Christian Skibakk Date: Fri, 20 Jul 2018 14:13:21 +0200 Subject: [PATCH 14/31] Fix pause/resume of ClientKeepAliveMonitor --- Source/MQTTnet/Server/MqttClientSession.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 429fdb45..7879ea6a 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -405,12 +405,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(); } } } From 2b10200101c8f8a985722fe152496259e687e384 Mon Sep 17 00:00:00 2001 From: Jens-Christian Skibakk Date: Sun, 22 Jul 2018 20:09:37 +0200 Subject: [PATCH 15/31] Small fix for inconsistent client name in log --- Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) 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); } } } From 043cfdb3f9e23c6cb29b31064bd44afef7d46e05 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Mon, 23 Jul 2018 17:58:29 +0200 Subject: [PATCH 16/31] fixed deadlock and object disposed exception --- Source/MQTTnet/Server/MqttClientSession.cs | 43 +++++++++++++-------- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 40 +++++++++++++++++++ 2 files changed, 67 insertions(+), 16 deletions(-) diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 8ef401c2..8ddb8455 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -126,18 +126,11 @@ namespace MQTTnet.Server _logger.Error(exception, "Client '{0}': Unhandled exception while receiving client packets.", ClientId); } - Stop(MqttClientDisconnectType.NotClean); + Stop(MqttClientDisconnectType.NotClean, true); } finally { - if (_adapter != null) - { - _adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted; - _adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted; - await Cleanup().ConfigureAwait(false); - } - - _adapter = null; + await Cleanup().ConfigureAwait(false); _cleanupHandle?.Dispose(); _cleanupHandle = null; @@ -153,8 +146,18 @@ namespace MQTTnet.Server { try { - await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); - _adapter.Dispose(); + var adapter = _adapter; + if (adapter == null) + { + return; + } + + _adapter = null; + + adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted; + adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted; + await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); + adapter.Dispose(); } catch (Exception exception) { @@ -163,6 +166,11 @@ namespace MQTTnet.Server } public void Stop(MqttClientDisconnectType type) + { + Stop(type, false); + } + + private void Stop(MqttClientDisconnectType type, bool isInsideSession) { try { @@ -183,7 +191,10 @@ namespace MQTTnet.Server _willMessage = null; - _workerTask?.GetAwaiter().GetResult(); + if (!isInsideSession) + { + _workerTask?.GetAwaiter().GetResult(); + } } finally { @@ -329,18 +340,18 @@ namespace MQTTnet.Server if (packet is MqttDisconnectPacket) { - Stop(MqttClientDisconnectType.Clean); + Stop(MqttClientDisconnectType.Clean, true); return; } if (packet is MqttConnectPacket) { - Stop(MqttClientDisconnectType.NotClean); + Stop(MqttClientDisconnectType.NotClean, true); return; } _logger.Warning(null, "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); - Stop(MqttClientDisconnectType.NotClean); + Stop(MqttClientDisconnectType.NotClean, true); } private void EnqueueSubscribedRetainedMessages(ICollection topicFilters) @@ -359,7 +370,7 @@ namespace MQTTnet.Server if (subscribeResult.CloseConnection) { - Stop(MqttClientDisconnectType.NotClean); + Stop(MqttClientDisconnectType.NotClean, true); return; } diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 117e6ef9..03888668 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -262,6 +262,46 @@ namespace MQTTnet.Core.Tests Assert.AreEqual(1, disconnectCalled); } + [TestMethod] + public async Task MqttServer_HandleCleanDisconnect() + { + MqttNetGlobalLogger.LogMessagePublished += (_, e) => + { + System.Diagnostics.Debug.WriteLine($"[{e.TraceMessage.Timestamp:s}] {e.TraceMessage.Source} {e.TraceMessage.Message}"); + }; + + var serverAdapter = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger()); + var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger()); + + var clientConnectedCalled = 0; + var clientDisconnectedCalled = 0; + + s.ClientConnected += (_, __) => clientConnectedCalled++; + s.ClientDisconnected += (_, __) => clientDisconnectedCalled++; + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost") + .Build(); + + await s.StartAsync(new MqttServerOptions()); + + var c1 = new MqttFactory().CreateMqttClient(); + + await c1.ConnectAsync(clientOptions); + + await Task.Delay(100); + + await c1.DisconnectAsync(); + + await Task.Delay(100); + + await s.StopAsync(); + + await Task.Delay(100); + + Assert.AreEqual(clientConnectedCalled, clientDisconnectedCalled); + } + [TestMethod] public async Task MqttServer_RetainedMessagesFlow() { From 603639126e2c95a763eb63b0b1f655ecee66a8d8 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Tue, 24 Jul 2018 13:49:23 +0200 Subject: [PATCH 17/31] Fix issue in processing of server messages. --- .../Server/MqttClientSessionsManager.cs | 83 +++++++++++-------- 1 file changed, 49 insertions(+), 34 deletions(-) diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 34080a00..48732a6d 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() @@ -133,54 +133,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) From 6b9015a9287a85182958339ae02cd18582f0be54 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 25 Jul 2018 20:18:27 +0200 Subject: [PATCH 18/31] Fix order of ClientConnected and ClientDisconnected events. --- Source/MQTTnet/Server/IMqttClientSession.cs | 2 +- Source/MQTTnet/Server/MqttClientSession.cs | 6 +-- .../Server/MqttClientSessionsManager.cs | 8 ++-- Tests/MQTTnet.Core.Tests/MqttClientTests.cs | 7 ++- .../MqttKeepAliveMonitorTests.cs | 4 +- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 48 +++++++++++++++++++ 6 files changed, 62 insertions(+), 13 deletions(-) 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/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 7879ea6a..fa90482b 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -65,7 +65,7 @@ namespace MQTTnet.Server status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived; } - public async Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) + public async Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) { if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket)); if (adapter == null) throw new ArgumentNullException(nameof(adapter)); @@ -129,8 +129,6 @@ namespace MQTTnet.Server _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; } - - return _wasCleanDisconnect; } public void Stop(MqttClientDisconnectType type) @@ -157,6 +155,8 @@ namespace MQTTnet.Server finally { _logger.Info("Client '{0}': Session stopped.", ClientId); + + _sessionsManager.Server.OnClientDisconnected(ClientId, _wasCleanDisconnect); } } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 48732a6d..c139151f 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -125,6 +125,7 @@ namespace MQTTnet.Server { _sessions.Remove(clientId); } + _logger.Verbose("Session for client '{0}' deleted.", clientId); } @@ -207,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); @@ -253,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) { @@ -278,8 +278,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 1334a803..f629f8f3 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs @@ -1,15 +1,18 @@ -using System.Net.Sockets; +using System; +using System.Net.Sockets; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; +using MQTTnet.Diagnostics; using MQTTnet.Exceptions; +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 95bb67da..58b27ba0 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -484,6 +484,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(); From 60bc4ebcee895c4cbf4970a8aa6a0e3c37be3c81 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 26 Jul 2018 10:35:52 +0200 Subject: [PATCH 19/31] Refactor proxy usage and support. Refactor TLS parameter usage. --- .../Client/MqttClientOptionsBuilder.cs | 93 +++++++------------ .../MqttClientOptionsBuilderTlsParameters.cs | 29 ++++++ .../Client/MqttClientWebSocketOptions.cs | 4 +- .../Client/MqttClientWebSocketProxyOptions.cs | 8 +- .../Implementations/MqttWebSocketChannel.cs | 50 +++++----- 5 files changed, 96 insertions(+), 88 deletions(-) create mode 100644 Source/MQTTnet/Client/MqttClientOptionsBuilderTlsParameters.cs diff --git a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs index ed8ae03d..c03e53ae 100644 --- a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs @@ -1,8 +1,5 @@ using System; using System.Linq; -using System.Net.Security; -using System.Security.Authentication; -using System.Security.Cryptography.X509Certificates; using MQTTnet.Serializer; namespace MQTTnet.Client @@ -10,10 +7,10 @@ namespace MQTTnet.Client public class MqttClientOptionsBuilder { private readonly MqttClientOptions _options = new MqttClientOptions(); + private MqttClientTcpOptions _tcpOptions; private MqttClientWebSocketOptions _webSocketOptions; - - private MqttClientTlsOptions _tlsOptions; + private MqttClientOptionsBuilderTlsParameters _tlsParameters; public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value) { @@ -88,7 +85,7 @@ namespace MQTTnet.Client throw new InvalidOperationException("A WebSocket channel must be set if MqttClientWebSocketProxy is configured."); } - _webSocketOptions.MqttClientWebSocketProxy = new MqttClientWebSocketProxyOptions + _webSocketOptions.ProxyOptions = new MqttClientWebSocketProxyOptions { Address = address, Username = username, @@ -112,86 +109,66 @@ namespace MQTTnet.Client return this; } + public MqttClientOptionsBuilder WithTls(MqttClientOptionsBuilderTlsParameters parameters) + { + _tlsParameters = parameters ?? throw new ArgumentNullException(nameof(parameters)); + return this; + } public MqttClientOptionsBuilder WithTls() { - return WithTls(null); - } - - public MqttClientOptionsBuilder WithTls(Func certificateValidationCallback) - { - return WithTls(SslProtocols.None, certificateValidationCallback); - } - - public MqttClientOptionsBuilder WithTls(SslProtocols sslProtocol, - Func certificateValidationCallback = null) - { - return WithTls(new byte[][] { }, sslProtocol, certificateValidationCallback); - } - - public MqttClientOptionsBuilder WithTls(byte[][] certificates, - SslProtocols sslProtocol = SslProtocols.Tls12, - Func certificateValidationCallback = null) - { - return WithTls(false, certificates, sslProtocol, certificateValidationCallback); - } - - public MqttClientOptionsBuilder WithTls(bool ignoreCertificateRevocationErrors, - byte[][] certificates = null, - SslProtocols sslProtocol = SslProtocols.Tls12, - Func certificateValidationCallback = null) - { - return WithTls(false, ignoreCertificateRevocationErrors, certificates, sslProtocol, certificateValidationCallback); - } - - public MqttClientOptionsBuilder WithTls(bool ignoreCertificateChainErrors, - bool ignoreCertificateRevocationErrors = false, - byte[][] certificates = null, - SslProtocols sslProtocol = SslProtocols.Tls12, - Func certificateValidationCallback = null) - { - return WithTls(false, ignoreCertificateChainErrors, ignoreCertificateRevocationErrors, certificates, sslProtocol, certificateValidationCallback); + return WithTls(new MqttClientOptionsBuilderTlsParameters { UseTls = true }); } + [Obsolete("Use method _WithTlps_ which accepts the _MqttClientOptionsBuilderTlsParameters_.")] public MqttClientOptionsBuilder WithTls( - bool allowUntrustedCertificates, + bool allowUntrustedCertificates = false, bool ignoreCertificateChainErrors = false, bool ignoreCertificateRevocationErrors = false, - byte[][] certificates = null, - SslProtocols sslProtocol = SslProtocols.Tls12, - Func certificateValidationCallback = null) + params byte[][] certificates) { - _tlsOptions = new MqttClientTlsOptions + _tlsParameters = new MqttClientOptionsBuilderTlsParameters { UseTls = true, AllowUntrustedCertificates = allowUntrustedCertificates, IgnoreCertificateChainErrors = ignoreCertificateChainErrors, IgnoreCertificateRevocationErrors = ignoreCertificateRevocationErrors, - Certificates = certificates?.ToList(), - SslProtocol = sslProtocol, - CertificateValidationCallback = certificateValidationCallback + Certificates = certificates?.ToList() }; return this; } - public IMqttClientOptions Build() { - if (_tlsOptions != null) + if (_tlsParameters != null) { if (_tcpOptions == null && _webSocketOptions == null) { throw new InvalidOperationException("A channel (TCP or WebSocket) must be set if TLS is configured."); } - if (_tcpOptions != null) + if (_tlsParameters?.UseTls == true) { - _tcpOptions.TlsOptions = _tlsOptions; - } - else if (_webSocketOptions != null) - { - _webSocketOptions.TlsOptions = _tlsOptions; + 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; + } } } 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/MqttClientWebSocketOptions.cs b/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs index 2cce4310..fe8f1d27 100644 --- a/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs +++ b/Source/MQTTnet/Client/MqttClientWebSocketOptions.cs @@ -13,9 +13,7 @@ namespace MQTTnet.Client public CookieContainer CookieContainer { get; set; } -#if NET452 || NET461 - public MqttClientWebSocketProxyOptions MqttClientWebSocketProxy { get; set; } = new MqttClientWebSocketProxyOptions(); -#endif + public MqttClientWebSocketProxyOptions ProxyOptions { get; set; } public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); diff --git a/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs b/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs index 81f48da8..0ac9df1f 100644 --- a/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs +++ b/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs @@ -1,6 +1,4 @@ -#if NET452 || NET461 - -namespace MQTTnet.Client +namespace MQTTnet.Client { public class MqttClientWebSocketProxyOptions { @@ -16,6 +14,4 @@ namespace MQTTnet.Client public string[] BypassList { get; set; } } -} - -#endif \ No newline at end of file +} \ No newline at end of file diff --git a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs index cf8c183a..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; @@ -6,10 +7,6 @@ using System.Threading.Tasks; using MQTTnet.Channel; using MQTTnet.Client; -#if NET452 || NET461 - using System.Net; -#endif - namespace MQTTnet.Implementations { public class MqttWebSocketChannel : IMqttChannel @@ -49,25 +46,10 @@ namespace MQTTnet.Implementations var clientWebSocket = new ClientWebSocket(); -#if NET452 || NET461 - // use proxy if we have an address - if (string.IsNullOrEmpty(_options.MqttClientWebSocketProxy.Address) == false) + if (_options.ProxyOptions != null) { - var proxyUri = new Uri(_options.MqttClientWebSocketProxy.Address); - - // use proxy credentials if we have them - if (string.IsNullOrEmpty(_options.MqttClientWebSocketProxy.Username) == false && string.IsNullOrEmpty(_options.MqttClientWebSocketProxy.Password) == false) - { - var credentials = new NetworkCredential(_options.MqttClientWebSocketProxy.Username, _options.MqttClientWebSocketProxy.Password, _options.MqttClientWebSocketProxy.Domain); - clientWebSocket.Options.Proxy = new WebProxy(proxyUri, _options.MqttClientWebSocketProxy.BypassOnLocal, _options.MqttClientWebSocketProxy.BypassList, credentials); - } - else - { - // use proxy without credentials - clientWebSocket.Options.Proxy = new WebProxy(proxyUri, _options.MqttClientWebSocketProxy.BypassOnLocal, _options.MqttClientWebSocketProxy.BypassList); - } + clientWebSocket.Options.Proxy = CreateProxy(); } -#endif if (_options.RequestHeaders != null) { @@ -156,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 From 0839fa575741e3fee1e96e7601ccdac90f41c140 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 26 Jul 2018 10:43:40 +0200 Subject: [PATCH 20/31] Update docs. --- Build/MQTTnet.nuspec | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) 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 From f70c79aaf9ab55a64dbeb4f0476a8c556b3e0c39 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 26 Jul 2018 11:34:24 +0200 Subject: [PATCH 21/31] Refactor options. --- README.md | 3 +- .../Client/MqttClientOptionsBuilder.cs | 33 ++++++++++--------- .../Client/MqttClientWebSocketProxyOptions.cs | 2 +- 3 files changed, 21 insertions(+), 17 deletions(-) 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 c03e53ae..85e2b4d8 100644 --- a/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/MqttClientOptionsBuilder.cs @@ -11,6 +11,7 @@ namespace MQTTnet.Client private MqttClientTcpOptions _tcpOptions; private MqttClientWebSocketOptions _webSocketOptions; private MqttClientOptionsBuilderTlsParameters _tlsParameters; + private MqttClientWebSocketProxyOptions _proxyOptions; public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value) { @@ -76,16 +77,9 @@ namespace MQTTnet.Client return this; } -#if NET452 || NET461 - public MqttClientOptionsBuilder WithProxy(string address, string username = null, string password = null, string domain = null, bool bypassOnLocal = false, string[] bypassList = null) { - if (_webSocketOptions == null) - { - throw new InvalidOperationException("A WebSocket channel must be set if MqttClientWebSocketProxy is configured."); - } - - _webSocketOptions.ProxyOptions = new MqttClientWebSocketProxyOptions + _proxyOptions = new MqttClientWebSocketProxyOptions { Address = address, Username = username, @@ -97,7 +91,6 @@ namespace MQTTnet.Client return this; } -#endif public MqttClientOptionsBuilder WithWebSocketServer(string uri) { @@ -120,7 +113,7 @@ namespace MQTTnet.Client return WithTls(new MqttClientOptionsBuilderTlsParameters { UseTls = true }); } - [Obsolete("Use method _WithTlps_ which accepts the _MqttClientOptionsBuilderTlsParameters_.")] + [Obsolete("Use method _WithTls_ which accepts the _MqttClientOptionsBuilderTlsParameters_.")] public MqttClientOptionsBuilder WithTls( bool allowUntrustedCertificates = false, bool ignoreCertificateChainErrors = false, @@ -141,13 +134,13 @@ namespace MQTTnet.Client public IMqttClientOptions Build() { + if (_tcpOptions == null && _webSocketOptions == null) + { + throw new InvalidOperationException("A channel must be set."); + } + if (_tlsParameters != null) { - if (_tcpOptions == null && _webSocketOptions == null) - { - throw new InvalidOperationException("A channel (TCP or WebSocket) must be set if TLS is configured."); - } - if (_tlsParameters?.UseTls == true) { var tlsOptions = new MqttClientTlsOptions @@ -172,6 +165,16 @@ namespace MQTTnet.Client } } + if (_proxyOptions != null) + { + if (_webSocketOptions == null) + { + throw new InvalidOperationException("Proxies are only supported for WebSocket connections."); + } + + _webSocketOptions.ProxyOptions = _proxyOptions; + } + _options.ChannelOptions = (IMqttClientChannelOptions)_tcpOptions ?? _webSocketOptions; return _options; diff --git a/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs b/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs index 0ac9df1f..13b13667 100644 --- a/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs +++ b/Source/MQTTnet/Client/MqttClientWebSocketProxyOptions.cs @@ -10,7 +10,7 @@ public string Domain { get; set; } - public bool BypassOnLocal { get; set; } = true; + public bool BypassOnLocal { get; set; } public string[] BypassList { get; set; } } From 2ced92100850c98deb24e7bc176dcf11b13c1ce2 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 26 Jul 2018 14:48:40 +0200 Subject: [PATCH 22/31] Refactoring. --- Source/MQTTnet/Server/MqttClientSession.cs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index ffc9fbdb..fe586636 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -29,7 +29,7 @@ namespace MQTTnet.Server private MqttApplicationMessage _willMessage; private bool _wasCleanDisconnect; private IMqttChannelAdapter _adapter; - private Task _workerTask; + private Task _workerTask; private IDisposable _cleanupHandle; public MqttClientSession( @@ -89,6 +89,7 @@ namespace MQTTnet.Server //woraround for https://github.com/dotnet/corefx/issues/24430 _cleanupHandle = _cancellationTokenSource.Token.Register(() => Cleanup()); //endworkaround + _wasCleanDisconnect = false; _willMessage = connectPacket.WillMessage; From 61b324043a19cb024e4b43d3b82dad102da66f7f Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 26 Jul 2018 20:02:11 +0200 Subject: [PATCH 23/31] Refactoring --- Build/MQTTnet.nuspec | 2 ++ Source/MQTTnet/Server/MqttClientSession.cs | 17 ++++++++++++----- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 10da9e58..1433105e 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -13,9 +13,11 @@ * [Core] Performance optimizations. * [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). +* [Client] Fixed a deadlock when clients are disconnecting (thanks to @JanEggers). * [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). +* [Server] Fixed a deadlock when clients are disconnecting (thanks to @JanEggers). 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 diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index fe586636..de3a7a48 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -81,13 +81,16 @@ namespace MQTTnet.Server try { _adapter = adapter; + adapter.ReadingPacketStarted += OnAdapterReadingPacketStarted; adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted; _cancellationTokenSource = new CancellationTokenSource(); //woraround for https://github.com/dotnet/corefx/issues/24430 - _cleanupHandle = _cancellationTokenSource.Token.Register(() => Cleanup()); +#pragma warning disable 4014 + _cleanupHandle = _cancellationTokenSource.Token.Register(() => CleanupAsync()); +#pragma warning restore 4014 //endworkaround _wasCleanDisconnect = false; @@ -131,7 +134,7 @@ namespace MQTTnet.Server } finally { - await Cleanup().ConfigureAwait(false); + await CleanupAsync().ConfigureAwait(false); _cleanupHandle?.Dispose(); _cleanupHandle = null; @@ -141,11 +144,11 @@ namespace MQTTnet.Server } } - private async Task Cleanup() + private async Task CleanupAsync() { + var adapter = _adapter; try { - var adapter = _adapter; if (adapter == null) { return; @@ -155,13 +158,17 @@ namespace MQTTnet.Server adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted; adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted; + await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); - adapter.Dispose(); } catch (Exception exception) { _logger.Error(exception, exception.Message); } + finally + { + adapter?.Dispose(); + } } public void Stop(MqttClientDisconnectType type) From b4f90decc74f74c04100ade1cf9676165b80f42d Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 27 Jul 2018 17:20:19 +0200 Subject: [PATCH 24/31] Fix unit tests to allow release builds. --- Source/MQTTnet/Properties/AssemblyInfo.cs | 2 ++ Tests/MQTTnet.Core.Tests/MqttClientTests.cs | 8 ++++---- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/Source/MQTTnet/Properties/AssemblyInfo.cs b/Source/MQTTnet/Properties/AssemblyInfo.cs index 8259cc15..1e212227 100644 --- a/Source/MQTTnet/Properties/AssemblyInfo.cs +++ b/Source/MQTTnet/Properties/AssemblyInfo.cs @@ -1,3 +1,5 @@ using System.Runtime.CompilerServices; +#if DEBUG [assembly:InternalsVisibleTo("MQTTnet.Core.Tests")] +#endif diff --git a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs index be899c31..212e1abb 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs @@ -8,8 +8,6 @@ using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Packets; -using MQTTnet.Implementations; -using MQTTnet.Server; namespace MQTTnet.Core.Tests { @@ -41,6 +39,7 @@ namespace MQTTnet.Core.Tests Assert.IsInstanceOfType(ex.InnerException, typeof(SocketException)); } +#if DEBUG [TestMethod] public async Task ClientCleanupOnAuthentificationFails() { @@ -55,11 +54,11 @@ namespace MQTTnet.Core.Tests }); - + var fake = new TestMqttCommunicationAdapterFactory(channel); var client = new MqttClient(fake, new MqttNetLogger()); - + try { await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("any-server").Build()); @@ -72,5 +71,6 @@ namespace MQTTnet.Core.Tests Assert.IsTrue(client._packetReceiverTask == null || client._packetReceiverTask.IsCompleted, "receive loop not completed"); Assert.IsTrue(client._keepAliveMessageSenderTask == null || client._keepAliveMessageSenderTask.IsCompleted, "keepalive loop not completed"); } +#endif } } From 80fbb2f0b231fd58c407ee020d5ce95f60025782 Mon Sep 17 00:00:00 2001 From: Christoph Stichlberger Date: Thu, 2 Aug 2018 16:25:14 +0200 Subject: [PATCH 25/31] Prevent unsubscription overtaking later subscription --- .../ManagedMqttClient.cs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 2eabce33..a38bcc27 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -306,15 +306,15 @@ namespace MQTTnet.Extensions.ManagedClient try { - if (subscriptions.Any()) - { - await _mqttClient.SubscribeAsync(subscriptions).ConfigureAwait(false); - } - if (unsubscriptions.Any()) { await _mqttClient.UnsubscribeAsync(unsubscriptions).ConfigureAwait(false); } + + if (subscriptions.Any()) + { + await _mqttClient.SubscribeAsync(subscriptions).ConfigureAwait(false); + } } catch (Exception exception) { From 6c08b5caf32dd28cf7f0a25f45b921e0594e25ad Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 4 Aug 2018 12:30:08 +0200 Subject: [PATCH 26/31] Update docs. --- Build/MQTTnet.AspNetCore.nuspec | 5 ++--- Build/MQTTnet.Extensions.ManagedClient.nuspec | 8 +++++--- Build/MQTTnet.Extensions.Rpc.nuspec | 8 +++++--- Build/MQTTnet.nuspec | 1 + 4 files changed, 13 insertions(+), 9 deletions(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index 43f056c6..5be2f945 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -10,13 +10,12 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is a support library to integrate MQTTnet into AspNetCore. - * Updated to MQTTnet 2.8.1. -* For more release notes please check the MQTTnet release notes. + * For more release notes please check the MQTTnet release notes. 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 - + diff --git a/Build/MQTTnet.Extensions.ManagedClient.nuspec b/Build/MQTTnet.Extensions.ManagedClient.nuspec index 44114a5d..99a58c4a 100644 --- a/Build/MQTTnet.Extensions.ManagedClient.nuspec +++ b/Build/MQTTnet.Extensions.ManagedClient.nuspec @@ -10,13 +10,12 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is an extension library which provides a managed MQTT client with additional features using MQTTnet. - * Updated to MQTTnet 2.8.1. -* For more release notes please check the MQTTnet release notes. + * For more release notes please check the MQTTnet release notes. 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 - + @@ -34,6 +33,9 @@ + + + \ No newline at end of file diff --git a/Build/MQTTnet.Extensions.Rpc.nuspec b/Build/MQTTnet.Extensions.Rpc.nuspec index 1c8bee74..736f92da 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -10,13 +10,12 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is an extension library which allows executing synchronous device calls including a response using MQTTnet. - * Updated to MQTTnet 2.8.1. -* For more release notes please check the MQTTnet release notes. + * For more release notes please check the MQTTnet release notes. 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 - + @@ -35,5 +34,8 @@ + + + \ No newline at end of file diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 1433105e..b92aece6 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -14,6 +14,7 @@ * [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). * [Client] Fixed a deadlock when clients are disconnecting (thanks to @JanEggers). +* [ManagedClient] Unsubscriptions are now pushed before subscriptions to support re-subscribing (thanks to @cstichlberger). * [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). From 0b5a751c5658c17a70f8983adb1912d0e3a84b61 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 15 Aug 2018 17:48:28 +0200 Subject: [PATCH 27/31] Add all methods of the factory to the factory interfaces. --- Source/MQTTnet/Client/IMqttClientFactory.cs | 4 ++++ Source/MQTTnet/Server/IMqttServerFactory.cs | 2 ++ 2 files changed, 6 insertions(+) diff --git a/Source/MQTTnet/Client/IMqttClientFactory.cs b/Source/MQTTnet/Client/IMqttClientFactory.cs index a15b499b..cd687afc 100644 --- a/Source/MQTTnet/Client/IMqttClientFactory.cs +++ b/Source/MQTTnet/Client/IMqttClientFactory.cs @@ -7,5 +7,9 @@ namespace MQTTnet.Client IMqttClient CreateMqttClient(); IMqttClient CreateMqttClient(IMqttNetLogger logger); + + IMqttClient CreateMqttClient(IMqttClientAdapterFactory adapterFactory); + + IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory adapterFactory); } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/IMqttServerFactory.cs b/Source/MQTTnet/Server/IMqttServerFactory.cs index 6e399cae..7d23ed36 100644 --- a/Source/MQTTnet/Server/IMqttServerFactory.cs +++ b/Source/MQTTnet/Server/IMqttServerFactory.cs @@ -8,6 +8,8 @@ namespace MQTTnet.Server { IMqttServer CreateMqttServer(); + IMqttServer CreateMqttServer(IMqttNetLogger logger); + IMqttServer CreateMqttServer(IEnumerable adapters, IMqttNetLogger logger); } } \ No newline at end of file From 4cc9f12a03bf9359f0c2c094ec6807ab84367db2 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Tue, 21 Aug 2018 21:32:03 +0200 Subject: [PATCH 28/31] added AspNetMqttServerOptionsBuilder --- .../AspNetMqttServerOptionsBuilder.cs | 15 +++++++++++++++ .../ServiceCollectionExtensions.cs | 14 +++++++------- 2 files changed, 22 insertions(+), 7 deletions(-) create mode 100644 Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs diff --git a/Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs b/Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs new file mode 100644 index 00000000..e0ae3330 --- /dev/null +++ b/Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs @@ -0,0 +1,15 @@ +using MQTTnet.Server; +using System; + +namespace MQTTnet.AspNetCore +{ + public class AspNetMqttServerOptionsBuilder : MqttServerOptionsBuilder + { + public AspNetMqttServerOptionsBuilder(IServiceProvider serviceProvider) + { + ServiceProvider = serviceProvider; + } + + public IServiceProvider ServiceProvider { get; } + } +} diff --git a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs index b5d0440a..45a0df5c 100644 --- a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -21,13 +21,13 @@ namespace MQTTnet.AspNetCore return services; } - public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action configure) - { - var builder = new MqttServerOptionsBuilder(); - - configure(builder); - - services.AddSingleton(builder.Build()); + public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action configure) + { + services.AddSingleton(s => { + var builder = new AspNetMqttServerOptionsBuilder(s); + configure(builder); + return builder.Build(); + }); services.AddHostedMqttServer(); From 976c62d9c2def88e8220fa7b6af8ca93201c6fb0 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 25 Aug 2018 12:08:42 +0200 Subject: [PATCH 29/31] added back the old function and added new overload --- .../ServiceCollectionExtensions.cs | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs index 45a0df5c..d9d9fdb8 100644 --- a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -20,9 +20,22 @@ namespace MQTTnet.AspNetCore return services; } - - public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action configure) - { + + public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action configure) + { + services.AddSingleton(s => { + var builder = new MqttServerOptionsBuilder(); + configure(builder); + return builder.Build(); + }); + + services.AddHostedMqttServer(); + + return services; + } + + public static IServiceCollection AddHostedMqttServerWithServices(this IServiceCollection services, Action configure) + { services.AddSingleton(s => { var builder = new AspNetMqttServerOptionsBuilder(s); configure(builder); From 028a39ec25853fca020c0320f971fefa8b67e8d5 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 13 Sep 2018 09:19:15 +0200 Subject: [PATCH 30/31] Update docs. --- Build/MQTTnet.nuspec | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index b92aece6..c59abcac 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -10,15 +10,8 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png 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. -* [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). -* [Client] Fixed a deadlock when clients are disconnecting (thanks to @JanEggers). -* [ManagedClient] Unsubscriptions are now pushed before subscriptions to support re-subscribing (thanks to @cstichlberger). -* [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). -* [Server] Fixed a deadlock when clients are disconnecting (thanks to @JanEggers). + * [Core] Added all factory methods to the factory interface. +* [Server] Added a new overload for configuring the ASP.net integration (thanks to @JanEggers). 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 From 682298b69fa2057086bf7fb26de6d456c49856bf Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 13 Sep 2018 09:29:26 +0200 Subject: [PATCH 31/31] Fix issue in pending messages handling for managed client. --- Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index a38bcc27..2b28deb9 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -226,7 +226,7 @@ namespace MQTTnet.Extensions.ManagedClient { try { - while (!cancellationToken.IsCancellationRequested) + while (!cancellationToken.IsCancellationRequested && _mqttClient.IsConnected) { var message = _messageQueue.Take(cancellationToken); if (message == null) @@ -367,7 +367,6 @@ namespace MQTTnet.Extensions.ManagedClient } var cts = new CancellationTokenSource(); - _publishingCancellationToken = cts; Task.Factory.StartNew(() => PublishQueuedMessages(cts.Token), cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);