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 34e563bc..c59abcac 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -10,12 +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. -* [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). + * [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 @@ -56,6 +52,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.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..d9d9fdb8 100644 --- a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -20,14 +20,27 @@ namespace MQTTnet.AspNetCore return services; } - + public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action configure) { - var builder = new MqttServerOptionsBuilder(); + services.AddSingleton(s => { + var builder = new MqttServerOptionsBuilder(); + configure(builder); + return builder.Build(); + }); - configure(builder); - - services.AddSingleton(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); + return builder.Build(); + }); services.AddHostedMqttServer(); 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.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 2eabce33..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) @@ -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) { @@ -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); 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/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/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index f7093eb1..575cc90f 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -25,10 +25,11 @@ 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; 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); + } } } @@ -351,7 +358,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 { @@ -395,7 +405,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/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 c48a06f6..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); @@ -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/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)); 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/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 diff --git a/Source/MQTTnet/Properties/AssemblyInfo.cs b/Source/MQTTnet/Properties/AssemblyInfo.cs new file mode 100644 index 00000000..1e212227 --- /dev/null +++ b/Source/MQTTnet/Properties/AssemblyInfo.cs @@ -0,0 +1,5 @@ +using System.Runtime.CompilerServices; + +#if DEBUG +[assembly:InternalsVisibleTo("MQTTnet.Core.Tests")] +#endif 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/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 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 95b15266..131cd397 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -29,6 +29,8 @@ namespace MQTTnet.Server private MqttApplicationMessage _willMessage; private bool _wasCleanDisconnect; private IMqttChannelAdapter _adapter; + private Task _workerTask; + private IDisposable _cleanupHandle; public MqttClientSession( string clientId, @@ -65,7 +67,13 @@ namespace MQTTnet.Server status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived; } - public async 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) { if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket)); if (adapter == null) throw new ArgumentNullException(nameof(adapter)); @@ -73,10 +81,18 @@ namespace MQTTnet.Server try { _adapter = adapter; + adapter.ReadingPacketStarted += OnAdapterReadingPacketStarted; adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted; _cancellationTokenSource = new CancellationTokenSource(); + + //woraround for https://github.com/dotnet/corefx/issues/24430 +#pragma warning disable 4014 + _cleanupHandle = _cancellationTokenSource.Token.Register(() => CleanupAsync()); +#pragma warning restore 4014 + //endworkaround + _wasCleanDisconnect = false; _willMessage = connectPacket.WillMessage; @@ -114,26 +130,55 @@ 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 CleanupAsync().ConfigureAwait(false); + _cleanupHandle?.Dispose(); + _cleanupHandle = null; + _adapter = null; - _cancellationTokenSource?.Cancel (); + _cancellationTokenSource?.Cancel(false); _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; } + } - return _wasCleanDisconnect; + private async Task CleanupAsync() + { + var adapter = _adapter; + try + { + if (adapter == null) + { + return; + } + + _adapter = null; + + adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted; + adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted; + + await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception exception) + { + _logger.Error(exception, exception.Message); + } + finally + { + adapter?.Dispose(); + } } public void Stop(MqttClientDisconnectType type) + { + Stop(type, false); + } + + private void Stop(MqttClientDisconnectType type, bool isInsideSession) { try { @@ -153,10 +198,17 @@ namespace MQTTnet.Server } _willMessage = null; + + if (!isInsideSession) + { + _workerTask?.GetAwaiter().GetResult(); + } } finally { _logger.Info("Client '{0}': Session stopped.", ClientId); + + _sessionsManager.Server.OnClientDisconnected(ClientId, _wasCleanDisconnect); } } @@ -300,18 +352,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) @@ -330,7 +382,7 @@ namespace MQTTnet.Server if (subscribeResult.CloseConnection) { - Stop(MqttClientDisconnectType.NotClean); + Stop(MqttClientDisconnectType.NotClean, true); return; } @@ -407,12 +459,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 34080a00..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) { @@ -249,22 +264,10 @@ 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); } - - Server.OnClientDisconnected(clientId, wasCleanDisconnect); } } 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/MqttClientTests.cs b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs index 1334a803..212e1abb 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs @@ -1,25 +1,29 @@ -using System.Net.Sockets; +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 { [TestClass] public class MqttClientTests { - [TestMethod] public async Task ClientDisconnectException() { 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 +33,44 @@ namespace MQTTnet.Core.Tests catch { } - - Assert.IsTrue(exceptionIsCorrect); + + Assert.IsNotNull(ex); + Assert.IsInstanceOfType(ex, typeof(MqttCommunicationException)); + Assert.IsInstanceOfType(ex.InnerException, typeof(SocketException)); } + +#if DEBUG + [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"); + } +#endif } } 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..0c34bdeb 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -233,6 +233,74 @@ 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(); + + var disconnectCalled = 0; + + await s.StartAsync(new MqttServerOptions()); + + var c1 = new MqttFactory().CreateMqttClient(); + c1.Disconnected += (sender, args) => disconnectCalled++; + + await c1.ConnectAsync(clientOptions); + + await Task.Delay(100); + + await s.StopAsync(); + + await Task.Delay(100); + + 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() @@ -484,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();