From daf22c762fc0161891fdcd5b6030ece88beecf84 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 23 Oct 2017 18:49:38 +0200 Subject: [PATCH 1/8] Small refactorings --- .../Server/MqttClientPendingMessagesQueue.cs | 4 +-- MQTTnet.Core/Server/MqttClientSession.cs | 28 +++++++++++-------- .../Server/MqttClientSessionsManager.cs | 1 + 3 files changed, 19 insertions(+), 14 deletions(-) diff --git a/MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs b/MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs index 551dc8e7..f16f689e 100644 --- a/MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs +++ b/MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs @@ -28,7 +28,7 @@ namespace MQTTnet.Core.Server { if (adapter == null) throw new ArgumentNullException(nameof(adapter)); - Task.Run(() => SendPendingPublishPacketsAsync(adapter, cancellationToken), cancellationToken); + Task.Factory.StartNew(async () => await SendPendingPublishPacketsAsync(adapter, cancellationToken), cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default).ConfigureAwait(false); } public void Enqueue(MqttPublishPacket publishPacket) @@ -74,7 +74,7 @@ namespace MQTTnet.Core.Server { _trace.Warning(nameof(MqttClientPendingMessagesQueue), exception, "Sending publish packet failed due to communication exception."); } - if (exception is OperationCanceledException) + else if (exception is OperationCanceledException) { } else diff --git a/MQTTnet.Core/Server/MqttClientSession.cs b/MQTTnet.Core/Server/MqttClientSession.cs index f913901e..5941a4fc 100644 --- a/MQTTnet.Core/Server/MqttClientSession.cs +++ b/MQTTnet.Core/Server/MqttClientSession.cs @@ -45,10 +45,9 @@ namespace MQTTnet.Core.Server { if (adapter == null) throw new ArgumentNullException(nameof(adapter)); - _willMessage = willMessage; - try { + _willMessage = willMessage; _adapter = adapter; _cancellationTokenSource = new CancellationTokenSource(); @@ -70,18 +69,23 @@ namespace MQTTnet.Core.Server public void Stop() { - if (_willMessage != null) + try { - _mqttClientSessionsManager.DispatchApplicationMessage(this, _willMessage); + _cancellationTokenSource?.Cancel(false); + _cancellationTokenSource?.Dispose(); + _cancellationTokenSource = null; + + _adapter = null; + + _trace.Information(nameof(MqttClientSession), "Client '{0}': Disconnected.", ClientId); + } + finally + { + if (_willMessage != null) + { + _mqttClientSessionsManager.DispatchApplicationMessage(this, _willMessage); + } } - - _cancellationTokenSource?.Cancel(false); - _cancellationTokenSource?.Dispose(); - _cancellationTokenSource = null; - - _adapter = null; - - _trace.Information(nameof(MqttClientSession), "Client '{0}': Disconnected.", ClientId); } public void EnqueuePublishPacket(MqttPublishPacket publishPacket) diff --git a/MQTTnet.Core/Server/MqttClientSessionsManager.cs b/MQTTnet.Core/Server/MqttClientSessionsManager.cs index aedd3d76..6238f2cb 100644 --- a/MQTTnet.Core/Server/MqttClientSessionsManager.cs +++ b/MQTTnet.Core/Server/MqttClientSessionsManager.cs @@ -160,6 +160,7 @@ namespace MQTTnet.Core.Server _clientSessions.Remove(connectPacket.ClientId); clientSession.Dispose(); clientSession = null; + _trace.Verbose(nameof(MqttClientSessionsManager), "Disposed existing session of client '{0}'.", connectPacket.ClientId); } else From 1c0b2b55b6157c17e7bfa59352a339021c4df60f Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 23 Oct 2017 18:57:36 +0200 Subject: [PATCH 2/8] Fix bugs --- MQTTnet.Core/Server/MqttClientSession.cs | 6 +++--- MQTTnet.Core/Server/MqttClientSessionsManager.cs | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/MQTTnet.Core/Server/MqttClientSession.cs b/MQTTnet.Core/Server/MqttClientSession.cs index 0870a662..2bd3617a 100644 --- a/MQTTnet.Core/Server/MqttClientSession.cs +++ b/MQTTnet.Core/Server/MqttClientSession.cs @@ -28,11 +28,11 @@ namespace MQTTnet.Core.Server public MqttClientSession(string clientId, MqttClientSessionsManager mqttClientSessionsManager, ILogger logger, ILogger msgQueueLogger) { - ClientId = clientId; - _options = mqttClientSessionsManager.Options; _mqttClientSessionsManager = mqttClientSessionsManager ?? throw new ArgumentNullException(nameof(mqttClientSessionsManager)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + ClientId = clientId; + _options = mqttClientSessionsManager.Options; _pendingMessagesQueue = new MqttClientPendingMessagesQueue(mqttClientSessionsManager.Options, this, msgQueueLogger); } @@ -78,7 +78,7 @@ namespace MQTTnet.Core.Server _adapter = null; - _trace.Information(nameof(MqttClientSession), "Client '{0}': Disconnected.", ClientId); + _logger.LogInformation("Client '{0}': Disconnected.", ClientId); } finally { diff --git a/MQTTnet.Core/Server/MqttClientSessionsManager.cs b/MQTTnet.Core/Server/MqttClientSessionsManager.cs index 5622f88d..b621b835 100644 --- a/MQTTnet.Core/Server/MqttClientSessionsManager.cs +++ b/MQTTnet.Core/Server/MqttClientSessionsManager.cs @@ -167,7 +167,7 @@ namespace MQTTnet.Core.Server clientSession.Dispose(); clientSession = null; - _trace.Verbose(nameof(MqttClientSessionsManager), "Disposed existing session of client '{0}'.", connectPacket.ClientId); + _logger.LogTrace("Disposed existing session of client '{0}'.", connectPacket.ClientId); } else { From a9b394401b0381557710b40a5b9b7078cceb79be Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 25 Oct 2017 19:50:37 +0200 Subject: [PATCH 3/8] Add support for chaning passwords --- .../Implementations/MqttServerAdapter.cs | 3 +- .../Implementations/MqttTcpChannel.cs | 2 +- .../MQTTnet.UniversalWindows/MqttFactory.cs | 2 +- MQTTnet.Core/Client/BaseMqttClientOptions.cs | 8 ++--- MQTTnet.Core/Client/IMqttClientCredentials.cs | 8 +++++ MQTTnet.Core/Client/IMqttClientOptions.cs | 8 ++--- MQTTnet.Core/Client/MqttClient.cs | 4 +-- MQTTnet.Core/Client/MqttClientCredentials.cs | 9 +++++ MQTTnet.Core/Client/MqttClientTcpOptions.cs | 2 +- .../Client/MqttClientWebSocketOptions.cs | 2 +- .../IManagedMqttClientOptions.cs | 2 -- .../ManagedClient/ManagedMqttClient.cs | 2 -- .../ManagedClient/ManagedMqttClientOptions.cs | 3 -- .../ManagedClientTest.cs | 33 ++++++++----------- .../MainPage.xaml.cs | 10 ++++-- 15 files changed, 52 insertions(+), 46 deletions(-) create mode 100644 MQTTnet.Core/Client/IMqttClientCredentials.cs create mode 100644 MQTTnet.Core/Client/MqttClientCredentials.cs diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs index 2d3db159..9e65715c 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs @@ -7,10 +7,8 @@ using System.Security.Cryptography.X509Certificates; using System.Threading; using System.Threading.Tasks; using MQTTnet.Core.Adapter; -using MQTTnet.Core.Serializer; using MQTTnet.Core.Server; using Microsoft.Extensions.Logging; -using MQTTnet.Core.Client; namespace MQTTnet.Implementations { @@ -18,6 +16,7 @@ namespace MQTTnet.Implementations { private readonly ILogger _logger; private readonly IMqttCommunicationAdapterFactory _mqttCommunicationAdapterFactory; + private CancellationTokenSource _cancellationTokenSource; private Socket _defaultEndpointSocket; private Socket _tlsEndpointSocket; diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs index 1a96f1e4..03206f1f 100644 --- a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs @@ -89,7 +89,7 @@ namespace MQTTnet.Implementations RawReceiveStream = ReceiveStream; } - private static Certificate LoadCertificate(BaseMqttClientOptions options) + private static Certificate LoadCertificate(MqttClientOptions options) { if (options.TlsOptions.Certificates == null || !options.TlsOptions.Certificates.Any()) { diff --git a/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs b/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs index 9761150e..a3aeca90 100644 --- a/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs +++ b/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs @@ -79,7 +79,7 @@ namespace MQTTnet public MqttPacketSerializer CreateSerializer(MqttProtocolVersion protocolVersion) { - return new MqttPacketSerializer() + return new MqttPacketSerializer { ProtocolVersion = protocolVersion }; diff --git a/MQTTnet.Core/Client/BaseMqttClientOptions.cs b/MQTTnet.Core/Client/BaseMqttClientOptions.cs index ca3f82e2..ea7ea7e1 100644 --- a/MQTTnet.Core/Client/BaseMqttClientOptions.cs +++ b/MQTTnet.Core/Client/BaseMqttClientOptions.cs @@ -3,20 +3,18 @@ using MQTTnet.Core.Serializer; namespace MQTTnet.Core.Client { - public abstract class BaseMqttClientOptions : IMqttClientOptions + public class MqttClientOptions : IMqttClientOptions { public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); public MqttApplicationMessage WillMessage { get; set; } - public string UserName { get; set; } - - public string Password { get; set; } - public string ClientId { get; set; } = Guid.NewGuid().ToString("N"); public bool CleanSession { get; set; } = true; + public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials(); + public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(5); public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10); diff --git a/MQTTnet.Core/Client/IMqttClientCredentials.cs b/MQTTnet.Core/Client/IMqttClientCredentials.cs new file mode 100644 index 00000000..c4dfc127 --- /dev/null +++ b/MQTTnet.Core/Client/IMqttClientCredentials.cs @@ -0,0 +1,8 @@ +namespace MQTTnet.Core.Client +{ + public interface IMqttClientCredentials + { + string Password { get; } + string Username { get; } + } +} \ No newline at end of file diff --git a/MQTTnet.Core/Client/IMqttClientOptions.cs b/MQTTnet.Core/Client/IMqttClientOptions.cs index 7e675ff8..f805de0b 100644 --- a/MQTTnet.Core/Client/IMqttClientOptions.cs +++ b/MQTTnet.Core/Client/IMqttClientOptions.cs @@ -5,14 +5,14 @@ namespace MQTTnet.Core.Client { public interface IMqttClientOptions { - bool CleanSession { get; } string ClientId { get; } + bool CleanSession { get; } + MqttApplicationMessage WillMessage { get; } + IMqttClientCredentials Credentials { get; } + TimeSpan DefaultCommunicationTimeout { get; } TimeSpan KeepAlivePeriod { get; } - string Password { get; } MqttProtocolVersion ProtocolVersion { get; } MqttClientTlsOptions TlsOptions { get; } - string UserName { get; } - MqttApplicationMessage WillMessage { get; } } } \ No newline at end of file diff --git a/MQTTnet.Core/Client/MqttClient.cs b/MQTTnet.Core/Client/MqttClient.cs index 67ffb126..e2094044 100644 --- a/MQTTnet.Core/Client/MqttClient.cs +++ b/MQTTnet.Core/Client/MqttClient.cs @@ -186,8 +186,8 @@ namespace MQTTnet.Core.Client var connectPacket = new MqttConnectPacket { ClientId = _options.ClientId, - Username = _options.UserName, - Password = _options.Password, + Username = _options.Credentials?.Username, + Password = _options.Credentials?.Password, CleanSession = _options.CleanSession, KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds, WillMessage = willApplicationMessage diff --git a/MQTTnet.Core/Client/MqttClientCredentials.cs b/MQTTnet.Core/Client/MqttClientCredentials.cs new file mode 100644 index 00000000..085778a3 --- /dev/null +++ b/MQTTnet.Core/Client/MqttClientCredentials.cs @@ -0,0 +1,9 @@ +namespace MQTTnet.Core.Client +{ + public class MqttClientCredentials : IMqttClientCredentials + { + public string Username { get; set; } + + public string Password { get; set; } + } +} diff --git a/MQTTnet.Core/Client/MqttClientTcpOptions.cs b/MQTTnet.Core/Client/MqttClientTcpOptions.cs index 51d2b973..beaa506a 100644 --- a/MQTTnet.Core/Client/MqttClientTcpOptions.cs +++ b/MQTTnet.Core/Client/MqttClientTcpOptions.cs @@ -1,6 +1,6 @@ namespace MQTTnet.Core.Client { - public class MqttClientTcpOptions : BaseMqttClientOptions + public class MqttClientTcpOptions : MqttClientOptions { public string Server { get; set; } diff --git a/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs b/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs index 522fb9bc..1a3fe03a 100644 --- a/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs +++ b/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs @@ -3,7 +3,7 @@ using System.Net; namespace MQTTnet.Core.Client { - public class MqttClientWebSocketOptions : BaseMqttClientOptions + public class MqttClientWebSocketOptions : MqttClientOptions { public string Uri { get; set; } diff --git a/MQTTnet.Core/ManagedClient/IManagedMqttClientOptions.cs b/MQTTnet.Core/ManagedClient/IManagedMqttClientOptions.cs index c812ea9c..298de9a1 100644 --- a/MQTTnet.Core/ManagedClient/IManagedMqttClientOptions.cs +++ b/MQTTnet.Core/ManagedClient/IManagedMqttClientOptions.cs @@ -10,7 +10,5 @@ namespace MQTTnet.Core.ManagedClient TimeSpan AutoReconnectDelay { get; } IManagedMqttClientStorage Storage { get; } - - Func PasswordProvider { get; } } } \ No newline at end of file diff --git a/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs b/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs index 0d37fe75..0018a58d 100644 --- a/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs +++ b/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs @@ -6,7 +6,6 @@ using System.Threading; using System.Threading.Tasks; using MQTTnet.Core.Client; using MQTTnet.Core.Exceptions; -using MQTTnet.Core.Packets; using MQTTnet.Core.Protocol; using Microsoft.Extensions.Logging; @@ -277,7 +276,6 @@ namespace MQTTnet.Core.ManagedClient try { - _options.PasswordProvider?.Invoke(_options); await _mqttClient.ConnectAsync(_options.ClientOptions).ConfigureAwait(false); return ReconnectionResult.Reconnected; } diff --git a/MQTTnet.Core/ManagedClient/ManagedMqttClientOptions.cs b/MQTTnet.Core/ManagedClient/ManagedMqttClientOptions.cs index 45b2af3c..2e24fd0a 100644 --- a/MQTTnet.Core/ManagedClient/ManagedMqttClientOptions.cs +++ b/MQTTnet.Core/ManagedClient/ManagedMqttClientOptions.cs @@ -10,8 +10,5 @@ namespace MQTTnet.Core.ManagedClient public TimeSpan AutoReconnectDelay { get; set; } = TimeSpan.FromSeconds(5); public IManagedMqttClientStorage Storage { get; set; } - - public Func PasswordProvider { get; set; } - } } diff --git a/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs b/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs index 05dd0cfa..587b847f 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs @@ -3,7 +3,6 @@ using System.Threading.Tasks; using MQTTnet.Core; using MQTTnet.Core.Client; using MQTTnet.Core.ManagedClient; -using MQTTnet.Core.Packets; using MQTTnet.Core.Protocol; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; @@ -25,32 +24,21 @@ namespace MQTTnet.TestApp.NetCore services.GetService() .AddConsole(); - ClientRetainedMessageHandler ms = new ClientRetainedMessageHandler(); - Func func = delegate (ManagedMqttClientOptions managedMqttClientOptions) - { - return "password"; - }; - + var ms = new ClientRetainedMessageHandler(); + var options = new ManagedMqttClientOptions { ClientOptions = new MqttClientTcpOptions { Server = "broker.hivemq.com", ClientId = "MQTTnetManagedClientTest", - Password = "pippo", + Credentials = new RandomPassword() }, AutoReconnectDelay = TimeSpan.FromSeconds(1), - Storage = ms, - PasswordProvider = o => - { - //o.ClientOptions.Password = GetPassword(); - return o.ClientOptions.Password; - } + Storage = ms }; - - try { var managedClient = services.GetRequiredService(); @@ -78,11 +66,18 @@ namespace MQTTnet.TestApp.NetCore } - public static string GetPassword() + public class RandomPassword : IMqttClientCredentials { - return "password"; - } + public string Password + { + get + { + return Guid.NewGuid().ToString(); // The random password. + } + } + public string Username => "the_static_user"; + } public class ClientRetainedMessageHandler : IManagedMqttClientStorage { diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs index 993e227e..68dd42a5 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs @@ -42,7 +42,7 @@ namespace MQTTnet.TestApp.UniversalWindows private async void Connect(object sender, RoutedEventArgs e) { - BaseMqttClientOptions options = null; + MqttClientOptions options = null; if (UseTcp.IsChecked == true) { options = new MqttClientTcpOptions @@ -64,8 +64,12 @@ namespace MQTTnet.TestApp.UniversalWindows throw new InvalidOperationException(); } - options.UserName = User.Text; - options.Password = Password.Text; + options.Credentials = new MqttClientCredentials + { + Username = User.Text, + Password = Password.Text + }; + options.ClientId = ClientId.Text; options.TlsOptions.UseTls = UseTls.IsChecked == true; options.TlsOptions.IgnoreCertificateChainErrors = true; From a18c53360354571520acce384f0d9a2602397b98 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 25 Oct 2017 22:33:07 +0200 Subject: [PATCH 4/8] Refactor options and add options builder --- .../ApplicationBuilderExtensions.cs | 1 - .../MqttWebSocketServerAdapter.cs | 2 +- .../ServiceCollectionExtensions.cs | 1 - .../Implementations/MqttTcpChannel.cs | 12 +- .../Implementations/MqttWebSocketChannel.cs | 7 +- Frameworks/MQTTnet.NetStandard/MqttFactory.cs | 16 +- .../Implementations/MqttTcpChannel.cs | 2 +- .../Implementations/MqttWebSocketChannel.cs | 9 +- .../MQTTnet.UniversalWindows/MqttFactory.cs | 4 +- .../Client/IApplicationMessagePublisher.cs | 10 ++ MQTTnet.Core/Client/IMqttClient.cs | 4 +- .../Client/IMqttClientChannelOptions.cs | 7 + MQTTnet.Core/Client/IMqttClientOptions.cs | 9 +- MQTTnet.Core/Client/MqttClient.cs | 16 +- ...tClientOptions.cs => MqttClientOptions.cs} | 6 +- .../Client/MqttClientOptionsBuilder.cs | 116 ++++++++++++ MQTTnet.Core/Client/MqttClientTcpOptions.cs | 4 +- MQTTnet.Core/Client/MqttClientTlsOptions.cs | 2 +- .../Client/MqttClientWebSocketOptions.cs | 4 +- .../ManagedClient/IManagedMqttClient.cs | 21 +++ .../ManagedClient/ManagedMqttClient.cs | 4 +- .../TestMqttServerAdapter.cs | 12 +- Tests/MQTTnet.TestApp.NetCore/ClientTest.cs | 99 +++++++++++ .../ManagedClientTest.cs | 9 +- .../PerformanceTest.cs | 17 +- Tests/MQTTnet.TestApp.NetCore/Program.cs | 121 ++----------- .../MainPage.xaml.cs | 166 ++++++++++-------- 27 files changed, 448 insertions(+), 233 deletions(-) create mode 100644 MQTTnet.Core/Client/IApplicationMessagePublisher.cs create mode 100644 MQTTnet.Core/Client/IMqttClientChannelOptions.cs rename MQTTnet.Core/Client/{BaseMqttClientOptions.cs => MqttClientOptions.cs} (77%) create mode 100644 MQTTnet.Core/Client/MqttClientOptionsBuilder.cs create mode 100644 MQTTnet.Core/ManagedClient/IManagedMqttClient.cs create mode 100644 Tests/MQTTnet.TestApp.NetCore/ClientTest.cs diff --git a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs index 87317a73..a62ecded 100644 --- a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs @@ -1,7 +1,6 @@ using System; using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.DependencyInjection; -using MQTTnet.AspNetCore; using MQTTnet.Core.Server; namespace MQTTnet.AspnetCore diff --git a/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs index 5cd5d555..abb30e37 100644 --- a/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs +++ b/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs @@ -8,7 +8,7 @@ using MQTTnet.Core.Channel; using MQTTnet.Core.Server; using MQTTnet.Implementations; -namespace MQTTnet.AspNetCore +namespace MQTTnet.AspnetCore { public class MqttWebSocketServerAdapter : IMqttServerAdapter, IDisposable { diff --git a/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs index b0297b21..1951368c 100644 --- a/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -1,6 +1,5 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; -using MQTTnet.AspNetCore; using MQTTnet.Core.Adapter; using MQTTnet.Core.Server; diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs index 7c051bfd..aac8a3c2 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs @@ -15,7 +15,7 @@ namespace MQTTnet.Implementations { private readonly MqttClientTcpOptions _options; -//todo: this can be used with min dependency NetStandard1.6 + //todo: this can be used with min dependency NetStandard1.6 #if NET45 // ReSharper disable once MemberCanBePrivate.Global // ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global @@ -24,7 +24,7 @@ namespace MQTTnet.Implementations private Socket _socket; private SslStream _sslStream; - + /// /// called on client sockets are created in connect /// @@ -48,7 +48,7 @@ namespace MQTTnet.Implementations public Stream SendStream { get; private set; } public Stream ReceiveStream { get; private set; } public Stream RawReceiveStream { get; private set; } - + public static Func CustomCertificateValidationCallback { get; set; } public async Task ConnectAsync() @@ -58,7 +58,7 @@ namespace MQTTnet.Implementations _socket = new Socket(SocketType.Stream, ProtocolType.Tcp); } -//todo: else brach can be used with min dependency NET46 + //todo: else brach can be used with min dependency NET46 #if NET451 await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false); #else @@ -120,7 +120,7 @@ namespace MQTTnet.Implementations return _options.TlsOptions.AllowUntrustedCertificates; } - private static X509CertificateCollection LoadCertificates(IMqttClientOptions options) + private static X509CertificateCollection LoadCertificates(MqttClientTcpOptions options) { var certificates = new X509CertificateCollection(); if (options.TlsOptions.Certificates == null) @@ -144,7 +144,7 @@ namespace MQTTnet.Implementations //cannot use this as default buffering prevents from receiving the first connect message //need two streams otherwise read and write have to be synchronized -//todo: if branch can be used with min dependency NetStandard1.6 + //todo: if branch can be used with min dependency NetStandard1.6 #if NET45 SendStream = new BufferedStream(RawReceiveStream, BufferSize); ReceiveStream = new BufferedStream(RawReceiveStream, BufferSize); diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs index 081c74a2..c120775b 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs @@ -32,7 +32,6 @@ namespace MQTTnet.Implementations } _webSocket = new ClientWebSocket(); - _webSocket.Options.KeepAliveInterval = _options.KeepAlivePeriod; if (_options.RequestHeaders != null) { @@ -77,12 +76,16 @@ namespace MQTTnet.Implementations return; } - await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false); + if (_webSocket.State == WebSocketState.Open || _webSocket.State == WebSocketState.Connecting) + { + await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false); + } } public void Dispose() { _webSocket?.Dispose(); + _webSocket = null; } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/MqttFactory.cs b/Frameworks/MQTTnet.NetStandard/MqttFactory.cs index 2c18ed96..b725f034 100644 --- a/Frameworks/MQTTnet.NetStandard/MqttFactory.cs +++ b/Frameworks/MQTTnet.NetStandard/MqttFactory.cs @@ -7,7 +7,6 @@ using Microsoft.Extensions.DependencyInjection; using MQTTnet.Implementations; using MQTTnet.Core.ManagedClient; using MQTTnet.Core.Server; -using System.Linq; using MQTTnet.Core.Channel; namespace MQTTnet @@ -18,7 +17,7 @@ namespace MQTTnet private static IServiceProvider BuildServiceProvider() { - var serviceProvider = new ServiceCollection() + var serviceProvider = new ServiceCollection() .AddMqttClient() .AddMqttServer() .AddLogging() @@ -40,10 +39,15 @@ namespace MQTTnet _serviceProvider = serviceProvider; } + public ILoggerFactory GetLoggerFactory() + { + return _serviceProvider.GetRequiredService(); + } + public IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options) { var logger = _serviceProvider.GetRequiredService>(); - return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options), CreateSerializer(options.ProtocolVersion), logger); + return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options.ChannelOptions), CreateSerializer(options.ProtocolVersion), logger); } public IMqttCommunicationAdapter CreateServerMqttCommunicationAdapter(IMqttCommunicationChannel channel) @@ -53,7 +57,7 @@ namespace MQTTnet return new MqttChannelCommunicationAdapter(channel, serializer, logger); } - public IMqttCommunicationChannel CreateMqttCommunicationChannel(IMqttClientOptions options) + public IMqttCommunicationChannel CreateMqttCommunicationChannel(IMqttClientChannelOptions options) { if (options == null) throw new ArgumentNullException(nameof(options)); @@ -66,7 +70,7 @@ namespace MQTTnet default: throw new NotSupportedException(); } - } + } public MqttTcpChannel CreateTcpChannel(MqttClientTcpOptions tcpOptions) { @@ -80,7 +84,7 @@ namespace MQTTnet public MqttPacketSerializer CreateSerializer(MqttProtocolVersion protocolVersion) { - return new MqttPacketSerializer() + return new MqttPacketSerializer { ProtocolVersion = protocolVersion }; diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs index 03206f1f..8fb45ab5 100644 --- a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs @@ -89,7 +89,7 @@ namespace MQTTnet.Implementations RawReceiveStream = ReceiveStream; } - private static Certificate LoadCertificate(MqttClientOptions options) + private static Certificate LoadCertificate(MqttClientTcpOptions options) { if (options.TlsOptions.Certificates == null || !options.TlsOptions.Certificates.Any()) { diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs index 48433165..6145802b 100644 --- a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs +++ b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs @@ -32,8 +32,7 @@ namespace MQTTnet.Implementations } _webSocket = new ClientWebSocket(); - _webSocket.Options.KeepAliveInterval = _options.KeepAlivePeriod; - + if (_options.RequestHeaders != null) { foreach (var requestHeader in _options.RequestHeaders) @@ -78,12 +77,16 @@ namespace MQTTnet.Implementations return; } - await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false); + if (_webSocket.State == WebSocketState.Open || _webSocket.State == WebSocketState.Connecting) + { + await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false); + } } public void Dispose() { _webSocket?.Dispose(); + _webSocket = null; } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs b/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs index a3aeca90..68fecd7f 100644 --- a/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs +++ b/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs @@ -42,7 +42,7 @@ namespace MQTTnet public IMqttCommunicationAdapter CreateClientMqttCommunicationAdapter(IMqttClientOptions options) { var logger = _serviceProvider.GetRequiredService>(); - return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options), CreateSerializer(options.ProtocolVersion), logger); + return new MqttChannelCommunicationAdapter(CreateMqttCommunicationChannel(options.ChannelOptions), CreateSerializer(options.ProtocolVersion), logger); } public IMqttCommunicationAdapter CreateServerMqttCommunicationAdapter(IMqttCommunicationChannel channel) @@ -52,7 +52,7 @@ namespace MQTTnet return new MqttChannelCommunicationAdapter(channel, serializer, logger); } - public IMqttCommunicationChannel CreateMqttCommunicationChannel(IMqttClientOptions options) + public IMqttCommunicationChannel CreateMqttCommunicationChannel(IMqttClientChannelOptions options) { if (options == null) throw new ArgumentNullException(nameof(options)); diff --git a/MQTTnet.Core/Client/IApplicationMessagePublisher.cs b/MQTTnet.Core/Client/IApplicationMessagePublisher.cs new file mode 100644 index 00000000..c77fd3f6 --- /dev/null +++ b/MQTTnet.Core/Client/IApplicationMessagePublisher.cs @@ -0,0 +1,10 @@ +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace MQTTnet.Core.Client +{ + public interface IApplicationMessagePublisher + { + Task PublishAsync(IEnumerable applicationMessages); + } +} diff --git a/MQTTnet.Core/Client/IMqttClient.cs b/MQTTnet.Core/Client/IMqttClient.cs index 9caefbda..6a39e211 100644 --- a/MQTTnet.Core/Client/IMqttClient.cs +++ b/MQTTnet.Core/Client/IMqttClient.cs @@ -5,7 +5,7 @@ using MQTTnet.Core.Packets; namespace MQTTnet.Core.Client { - public interface IMqttClient : IApplicationMessageReceiver + public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher { bool IsConnected { get; } @@ -17,7 +17,5 @@ namespace MQTTnet.Core.Client Task> SubscribeAsync(IEnumerable topicFilters); Task UnsubscribeAsync(IEnumerable topicFilters); - - Task PublishAsync(IEnumerable applicationMessages); } } \ No newline at end of file diff --git a/MQTTnet.Core/Client/IMqttClientChannelOptions.cs b/MQTTnet.Core/Client/IMqttClientChannelOptions.cs new file mode 100644 index 00000000..004430f9 --- /dev/null +++ b/MQTTnet.Core/Client/IMqttClientChannelOptions.cs @@ -0,0 +1,7 @@ +namespace MQTTnet.Core.Client +{ + public interface IMqttClientChannelOptions + { + MqttClientTlsOptions TlsOptions { get; } + } +} diff --git a/MQTTnet.Core/Client/IMqttClientOptions.cs b/MQTTnet.Core/Client/IMqttClientOptions.cs index f805de0b..d2bcf9d3 100644 --- a/MQTTnet.Core/Client/IMqttClientOptions.cs +++ b/MQTTnet.Core/Client/IMqttClientOptions.cs @@ -6,13 +6,14 @@ namespace MQTTnet.Core.Client public interface IMqttClientOptions { string ClientId { get; } + IMqttClientCredentials Credentials { get; } bool CleanSession { get; } MqttApplicationMessage WillMessage { get; } - IMqttClientCredentials Credentials { get; } - - TimeSpan DefaultCommunicationTimeout { get; } + + TimeSpan CommunicationTimeout { get; } TimeSpan KeepAlivePeriod { get; } MqttProtocolVersion ProtocolVersion { get; } - MqttClientTlsOptions TlsOptions { get; } + + IMqttClientChannelOptions ChannelOptions { get; } } } \ No newline at end of file diff --git a/MQTTnet.Core/Client/MqttClient.cs b/MQTTnet.Core/Client/MqttClient.cs index e2094044..e33b2581 100644 --- a/MQTTnet.Core/Client/MqttClient.cs +++ b/MQTTnet.Core/Client/MqttClient.cs @@ -42,6 +42,7 @@ namespace MQTTnet.Core.Client public async Task ConnectAsync(IMqttClientOptions options) { if (options == null) throw new ArgumentNullException(nameof(options)); + if (options.ChannelOptions == null) throw new ArgumentException("ChannelOptions are not set."); ThrowIfConnected("It is not allowed to connect with a server after the connection is established."); @@ -56,7 +57,7 @@ namespace MQTTnet.Core.Client _scopeHandle = _logger.BeginScope(options.ClientId); _logger.LogTrace("Trying to connect with server."); - await _adapter.ConnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false); + await _adapter.ConnectAsync(_options.CommunicationTimeout).ConfigureAwait(false); _logger.LogTrace("Connection with server established."); await SetupIncomingPacketProcessingAsync(); @@ -73,8 +74,9 @@ namespace MQTTnet.Core.Client Connected?.Invoke(this, new MqttClientConnectedEventArgs(connectResponse.IsSessionPresent)); return new MqttClientConnectResult(connectResponse.IsSessionPresent); } - catch (Exception) + catch (Exception exception) { + _logger.LogError(new EventId(), exception, "Error while connecting with server."); await DisconnectInternalAsync().ConfigureAwait(false); throw; } @@ -149,7 +151,7 @@ namespace MQTTnet.Core.Client case MqttQualityOfServiceLevel.AtMostOnce: { // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] - await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, qosPackets); + await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosPackets); break; } case MqttQualityOfServiceLevel.AtLeastOnce: @@ -247,7 +249,7 @@ namespace MQTTnet.Core.Client try { - await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false); + await _adapter.DisconnectAsync(_options.CommunicationTimeout).ConfigureAwait(false); _logger.LogInformation("Disconnected from adapter."); } catch (Exception exception) @@ -354,13 +356,13 @@ namespace MQTTnet.Core.Client private Task SendAsync(MqttBasePacket packet) { - return _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, packet); + return _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, packet); } private async Task SendAndReceiveAsync(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket { - var packetAwaiter = _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.DefaultCommunicationTimeout); - await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false); + var packetAwaiter = _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.CommunicationTimeout); + await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false); return (TResponsePacket)await packetAwaiter.ConfigureAwait(false); } diff --git a/MQTTnet.Core/Client/BaseMqttClientOptions.cs b/MQTTnet.Core/Client/MqttClientOptions.cs similarity index 77% rename from MQTTnet.Core/Client/BaseMqttClientOptions.cs rename to MQTTnet.Core/Client/MqttClientOptions.cs index ea7ea7e1..04cf8bdb 100644 --- a/MQTTnet.Core/Client/BaseMqttClientOptions.cs +++ b/MQTTnet.Core/Client/MqttClientOptions.cs @@ -5,8 +5,6 @@ namespace MQTTnet.Core.Client { public class MqttClientOptions : IMqttClientOptions { - public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); - public MqttApplicationMessage WillMessage { get; set; } public string ClientId { get; set; } = Guid.NewGuid().ToString("N"); @@ -17,8 +15,10 @@ namespace MQTTnet.Core.Client public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(5); - public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10); + public TimeSpan CommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10); public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; + + public IMqttClientChannelOptions ChannelOptions { get; set; } } } diff --git a/MQTTnet.Core/Client/MqttClientOptionsBuilder.cs b/MQTTnet.Core/Client/MqttClientOptionsBuilder.cs new file mode 100644 index 00000000..1eb2a347 --- /dev/null +++ b/MQTTnet.Core/Client/MqttClientOptionsBuilder.cs @@ -0,0 +1,116 @@ +using System; +using MQTTnet.Core.Serializer; + +namespace MQTTnet.Core.Client +{ + public class MqttClientOptionsBuilder + { + private readonly MqttClientOptions _options = new MqttClientOptions(); + private MqttClientTcpOptions _tcpOptions; + private MqttClientWebSocketOptions _webSocketOptions; + + private MqttClientTlsOptions _tlsOptions; + + public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion protocolVersion) + { + _options.ProtocolVersion = protocolVersion; + return this; + } + + public MqttClientOptionsBuilder WithCommunicationTimeout(TimeSpan communicationTimeout) + { + _options.CommunicationTimeout = communicationTimeout; + return this; + } + + public MqttClientOptionsBuilder WithCleanSession(bool value = true) + { + _options.CleanSession = value; + return this; + } + + public MqttClientOptionsBuilder WithKeepAlivePeriod(TimeSpan keepAlivePeriod) + { + _options.KeepAlivePeriod = keepAlivePeriod; + return this; + } + + public MqttClientOptionsBuilder WithClientId(string clientId) + { + _options.ClientId = clientId; + return this; + } + + public MqttClientOptionsBuilder WithWillMessage(MqttApplicationMessage applicationMessage) + { + _options.WillMessage = applicationMessage; + return this; + } + + public MqttClientOptionsBuilder WithCredentials(string username, string password = null) + { + _options.Credentials = new MqttClientCredentials + { + Username = username, + Password = password + }; + + return this; + } + + public MqttClientOptionsBuilder WithTcpServer(string server, int? port = null) + { + _tcpOptions = new MqttClientTcpOptions + { + Server = server, + Port = port + }; + + return this; + } + + public MqttClientOptionsBuilder WithWebSocketServer(string uri) + { + _webSocketOptions = new MqttClientWebSocketOptions + { + Uri = uri + }; + + return this; + } + + public MqttClientOptionsBuilder WithTls() + { + _tlsOptions = new MqttClientTlsOptions + { + UseTls = true + }; + + return this; + } + + public IMqttClientOptions Build() + { + if (_tlsOptions != null) + { + if (_tcpOptions == null && _webSocketOptions == null) + { + throw new InvalidOperationException("A channel (TCP or WebSocket) must be set if TLS is configured."); + } + + if (_tcpOptions != null) + { + _options.ChannelOptions = _tcpOptions; + } + else + { + _options.ChannelOptions = _webSocketOptions; + } + } + + _options.ChannelOptions = (IMqttClientChannelOptions)_tcpOptions ?? _webSocketOptions; + + return _options; + } + } +} diff --git a/MQTTnet.Core/Client/MqttClientTcpOptions.cs b/MQTTnet.Core/Client/MqttClientTcpOptions.cs index beaa506a..e5cf8d6c 100644 --- a/MQTTnet.Core/Client/MqttClientTcpOptions.cs +++ b/MQTTnet.Core/Client/MqttClientTcpOptions.cs @@ -1,9 +1,11 @@ namespace MQTTnet.Core.Client { - public class MqttClientTcpOptions : MqttClientOptions + public class MqttClientTcpOptions : IMqttClientChannelOptions { public string Server { get; set; } public int? Port { get; set; } + + public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); } } diff --git a/MQTTnet.Core/Client/MqttClientTlsOptions.cs b/MQTTnet.Core/Client/MqttClientTlsOptions.cs index 6fc4ab93..33cd691f 100644 --- a/MQTTnet.Core/Client/MqttClientTlsOptions.cs +++ b/MQTTnet.Core/Client/MqttClientTlsOptions.cs @@ -2,7 +2,7 @@ namespace MQTTnet.Core.Client { - public sealed class MqttClientTlsOptions + public class MqttClientTlsOptions { public bool UseTls { get; set; } diff --git a/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs b/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs index 1a3fe03a..25e4ae81 100644 --- a/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs +++ b/MQTTnet.Core/Client/MqttClientWebSocketOptions.cs @@ -3,7 +3,7 @@ using System.Net; namespace MQTTnet.Core.Client { - public class MqttClientWebSocketOptions : MqttClientOptions + public class MqttClientWebSocketOptions : IMqttClientChannelOptions { public string Uri { get; set; } @@ -12,5 +12,7 @@ namespace MQTTnet.Core.Client public ICollection SubProtocols { get; set; } public CookieContainer CookieContainer { get; set; } + + public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); } } diff --git a/MQTTnet.Core/ManagedClient/IManagedMqttClient.cs b/MQTTnet.Core/ManagedClient/IManagedMqttClient.cs new file mode 100644 index 00000000..8d2ac8cc --- /dev/null +++ b/MQTTnet.Core/ManagedClient/IManagedMqttClient.cs @@ -0,0 +1,21 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using MQTTnet.Core.Client; + +namespace MQTTnet.Core.ManagedClient +{ + public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher + { + bool IsConnected { get; } + + event EventHandler Connected; + event EventHandler Disconnected; + + Task StartAsync(IManagedMqttClientOptions options); + Task StopAsync(); + + Task SubscribeAsync(IEnumerable topicFilters); + Task UnsubscribeAsync(IEnumerable topicFilters); + } +} \ No newline at end of file diff --git a/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs b/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs index 0018a58d..61c0067a 100644 --- a/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs +++ b/MQTTnet.Core/ManagedClient/ManagedMqttClient.cs @@ -11,7 +11,7 @@ using Microsoft.Extensions.Logging; namespace MQTTnet.Core.ManagedClient { - public class ManagedMqttClient : IApplicationMessageReceiver + public class ManagedMqttClient : IManagedMqttClient { private readonly ManagedMqttClientStorageManager _storageManager = new ManagedMqttClientStorageManager(); private readonly BlockingCollection _messageQueue = new BlockingCollection(); @@ -91,7 +91,7 @@ namespace MQTTnet.Core.ManagedClient return Task.FromResult(0); } - public Task EnqueueAsync(IEnumerable applicationMessages) + public Task PublishAsync(IEnumerable applicationMessages) { if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); diff --git a/Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs b/Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs index 0655a096..067d08fd 100644 --- a/Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs +++ b/Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs @@ -21,7 +21,17 @@ namespace MQTTnet.Core.Tests var connected = WaitForClientToConnect(server, clientId); FireClientAcceptedEvent(adapterB); - await client.ConnectAsync(new MqttClientTcpOptions { ClientId = clientId, WillMessage = willMessage }); + + var options = new MqttClientOptions + { + ClientId = clientId, + WillMessage = willMessage, + ChannelOptions = new MqttClientTcpOptions() + }; + + options.ChannelOptions = new MqttClientTcpOptions(); + + await client.ConnectAsync(options); await connected; return client; diff --git a/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs b/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs new file mode 100644 index 00000000..cccdff66 --- /dev/null +++ b/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs @@ -0,0 +1,99 @@ +using System; +using System.Text; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using MQTTnet.Core; +using MQTTnet.Core.Client; + +namespace MQTTnet.TestApp.NetCore +{ + public static class ClientTest + { + public static async Task RunAsync() + { + try + { + var options = new MqttClientOptions + { + ClientId = "XYZ", + CleanSession = true, + ////ChannelOptions = new MqttClientTcpOptions + ////{ + //// Server = "localhost" + ////}, + ChannelOptions = new MqttClientWebSocketOptions + { + Uri = "localhost" + } + }; + + var factory = new MqttFactory(); + factory.GetLoggerFactory().AddConsole(); + + var client = factory.CreateMqttClient(); + + client.ApplicationMessageReceived += (s, e) => + { + Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###"); + Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}"); + Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}"); + Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}"); + Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}"); + Console.WriteLine(); + }; + + client.Connected += async (s, e) => + { + Console.WriteLine("### CONNECTED WITH SERVER ###"); + + await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("#").Build()); + + Console.WriteLine("### SUBSCRIBED ###"); + }; + + client.Disconnected += async (s, e) => + { + Console.WriteLine("### DISCONNECTED FROM SERVER ###"); + await Task.Delay(TimeSpan.FromSeconds(5)); + + try + { + await client.ConnectAsync(options); + } + catch + { + Console.WriteLine("### RECONNECTING FAILED ###"); + } + }; + + try + { + await client.ConnectAsync(options); + } + catch (Exception exception) + { + Console.WriteLine("### CONNECTING FAILED ###" + Environment.NewLine + exception); + } + + Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###"); + + while (true) + { + Console.ReadLine(); + + var applicationMessage = new MqttApplicationMessageBuilder() + .WithTopic("A/B/C") + .WithPayload("Hello World") + .WithAtLeastOnceQoS() + .Build(); + + await client.PublishAsync(applicationMessage); + } + } + catch (Exception exception) + { + Console.WriteLine(exception); + } + } + } +} diff --git a/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs b/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs index 587b847f..a02e72c4 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs @@ -28,11 +28,14 @@ namespace MQTTnet.TestApp.NetCore var options = new ManagedMqttClientOptions { - ClientOptions = new MqttClientTcpOptions + ClientOptions = new MqttClientOptions { - Server = "broker.hivemq.com", ClientId = "MQTTnetManagedClientTest", - Credentials = new RandomPassword() + Credentials = new RandomPassword(), + ChannelOptions = new MqttClientTcpOptions + { + Server = "broker.hivemq.com" + } }, AutoReconnectDelay = TimeSpan.FromSeconds(1), diff --git a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs index 746aa74e..6ef76682 100644 --- a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs @@ -2,7 +2,6 @@ using Microsoft.Extensions.Logging; using MQTTnet.Core; using MQTTnet.Core.Client; -using MQTTnet.Core.Packets; using MQTTnet.Core.Protocol; using MQTTnet.Core.Server; using System; @@ -20,7 +19,8 @@ namespace MQTTnet.TestApp.NetCore public static async Task RunAsync() { var services = new ServiceCollection() - .AddMqttServer(options => { + .AddMqttServer(options => + { options.ConnectionValidator = p => { @@ -41,11 +41,10 @@ namespace MQTTnet.TestApp.NetCore .AddLogging() .BuildServiceProvider(); - services.GetService() - .AddConsole(minLevel: LogLevel.Warning, includeScopes: true); - + services.GetService().AddConsole(LogLevel.Warning, true); + Console.WriteLine("Press 'c' for concurrent sends. Otherwise in one batch."); - var concurrent = Console.ReadKey(intercept: true).KeyChar == 'c'; + var concurrent = Console.ReadKey(true).KeyChar == 'c'; var server = Task.Factory.StartNew(() => RunServerAsync(services), TaskCreationOptions.LongRunning); var client = Task.Factory.StartNew(() => RunClientAsync(2000, TimeSpan.FromMilliseconds(10), services, concurrent), TaskCreationOptions.LongRunning); @@ -62,12 +61,12 @@ namespace MQTTnet.TestApp.NetCore { try { - var options = new MqttClientTcpOptions + var options = new MqttClientOptions { - Server = "localhost", + ChannelOptions = new MqttClientTcpOptions { Server = "localhost" }, ClientId = "Client1", CleanSession = true, - DefaultCommunicationTimeout = TimeSpan.FromMinutes(10) + CommunicationTimeout = TimeSpan.FromMinutes(10) }; var client = serviceProvider.GetRequiredService(); diff --git a/Tests/MQTTnet.TestApp.NetCore/Program.cs b/Tests/MQTTnet.TestApp.NetCore/Program.cs index f0b28c80..2e1d8c90 100644 --- a/Tests/MQTTnet.TestApp.NetCore/Program.cs +++ b/Tests/MQTTnet.TestApp.NetCore/Program.cs @@ -1,17 +1,12 @@ using MQTTnet.Core; using MQTTnet.Core.Client; -using MQTTnet.Core.Packets; -using MQTTnet.Core.Protocol; using MQTTnet.Core.Server; using System; using System.Collections.Generic; using System.IO; -using System.Security.Cryptography.X509Certificates; -using System.Text; using System.Threading; using System.Threading.Tasks; using Newtonsoft.Json; -using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; namespace MQTTnet.TestApp.NetCore @@ -25,11 +20,11 @@ namespace MQTTnet.TestApp.NetCore Console.WriteLine("2 = Start server"); Console.WriteLine("3 = Start performance test"); Console.WriteLine("4 = Start managed client"); - + var pressedKey = Console.ReadKey(true); if (pressedKey.KeyChar == '1') { - Task.Run(RunClientAsync); + Task.Run(ClientTest.RunAsync); } else if (pressedKey.KeyChar == '2') { @@ -46,105 +41,22 @@ namespace MQTTnet.TestApp.NetCore Thread.Sleep(Timeout.Infinite); } - - private static async Task RunClientAsync() - { - try - { - var services = new ServiceCollection() - .AddMqttServer() - .AddLogging() - .BuildServiceProvider(); - - services.GetService() - .AddConsole(); - - var options = new MqttClientWebSocketOptions - { - Uri = "localhost", - ClientId = "XYZ", - CleanSession = true - }; - - var client = services.GetRequiredService(); - client.ApplicationMessageReceived += (s, e) => - { - Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###"); - Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}"); - Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}"); - Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}"); - Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}"); - Console.WriteLine(); - }; - - client.Connected += async (s, e) => - { - Console.WriteLine("### CONNECTED WITH SERVER ###"); - - await client.SubscribeAsync(new List - { - new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce) - }); - - Console.WriteLine("### SUBSCRIBED ###"); - }; - - client.Disconnected += async (s, e) => - { - Console.WriteLine("### DISCONNECTED FROM SERVER ###"); - await Task.Delay(TimeSpan.FromSeconds(5)); - - try - { - await client.ConnectAsync(options); - } - catch - { - Console.WriteLine("### RECONNECTING FAILED ###"); - } - }; - - try - { - await client.ConnectAsync(options); - } - catch (Exception exception) - { - Console.WriteLine("### CONNECTING FAILED ###" + Environment.NewLine + exception); - } - - Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###"); - - while (true) - { - Console.ReadLine(); - - var applicationMessage = new MqttApplicationMessage - { - Topic = "A/B/C", - Payload = Encoding.UTF8.GetBytes("Hello World"), - QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce - }; - - await client.PublishAsync(applicationMessage); - } - } - catch (Exception exception) - { - Console.WriteLine(exception); - } - } - + + // This code is used at the Wiki on GitHub! // ReSharper disable once UnusedMember.Local private static async void WikiCode() { { - var serviceProvider = new ServiceCollection() - .AddMqttServer() - .AddLogging() - .BuildServiceProvider(); + var client = new MqttFactory().CreateMqttClient(); - var client = serviceProvider.GetRequiredService(); + var options = new MqttClientOptionsBuilder() + .WithClientId("Client1") + .WithTcpServer("broker.hivemq.com") + .WithCredentials("bud", "%spencer%") + .WithTls() + .Build(); + + await client.ConnectAsync(options); var message = new MqttApplicationMessageBuilder() .WithTopic("MyTopic") @@ -157,9 +69,10 @@ namespace MQTTnet.TestApp.NetCore } { - var message = new MqttApplicationMessageBuilder() - .WithTopic("/MQTTnet/is/awesome") - .Build(); + var factory = new MqttFactory(); + factory.GetLoggerFactory().AddConsole(); + + var client = factory.CreateMqttClient(); } } } diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs index 68dd42a5..3663ba4c 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs @@ -8,9 +8,7 @@ using Microsoft.Extensions.DependencyInjection; using MQTTnet.Core; using MQTTnet.Core.Client; using MQTTnet.Core.Diagnostics; -using MQTTnet.Core.Packets; using MQTTnet.Core.Protocol; -using MQTTnet.Core.Server; using MQTTnet.Implementations; namespace MQTTnet.TestApp.UniversalWindows @@ -42,24 +40,35 @@ namespace MQTTnet.TestApp.UniversalWindows private async void Connect(object sender, RoutedEventArgs e) { - MqttClientOptions options = null; + var tlsOptions = new MqttClientTlsOptions + { + UseTls = UseTls.IsChecked == true, + IgnoreCertificateChainErrors = true, + IgnoreCertificateRevocationErrors = true, + AllowUntrustedCertificates = true + }; + + var options = new MqttClientOptions { ClientId = ClientId.Text }; + if (UseTcp.IsChecked == true) { - options = new MqttClientTcpOptions + options.ChannelOptions = new MqttClientTcpOptions { - Server = Server.Text + Server = Server.Text, + TlsOptions = tlsOptions }; } if (UseWs.IsChecked == true) { - options = new MqttClientWebSocketOptions + options.ChannelOptions = new MqttClientWebSocketOptions { - Uri = Server.Text + Uri = Server.Text, + TlsOptions = tlsOptions }; } - if (options == null) + if (options.ChannelOptions == null) { throw new InvalidOperationException(); } @@ -70,12 +79,6 @@ namespace MQTTnet.TestApp.UniversalWindows Password = Password.Text }; - options.ClientId = ClientId.Text; - options.TlsOptions.UseTls = UseTls.IsChecked == true; - options.TlsOptions.IgnoreCertificateChainErrors = true; - options.TlsOptions.IgnoreCertificateRevocationErrors = true; - options.TlsOptions.AllowUntrustedCertificates = true; - try { if (_mqttClient != null) @@ -205,73 +208,94 @@ namespace MQTTnet.TestApp.UniversalWindows // ReSharper disable once UnusedMember.Local private async Task WikiCode() { - var services = new ServiceCollection() - .AddMqttClient() - .AddMqttServer(options => - { - options.ConnectionValidator = c => - { - if (c.ClientId.Length < 10) - { - return MqttConnectReturnCode.ConnectionRefusedIdentifierRejected; - } - - if (c.Username != "mySecretUser") - { - return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; - } - - if (c.Password != "mySecretPassword") - { - return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; - } - - return MqttConnectReturnCode.ConnectionAccepted; - }; - }) - .BuildServiceProvider(); - - var factory = new MqttFactory(services); - - var mqttClient = factory.CreateMqttClient(); - - // ---------------------------------- - - var tcpOptions = new MqttClientTcpOptions { - Server = "broker.hivemq.org", - ClientId = "TestClient" - }; + // Create a new MQTT client + var services = new ServiceCollection() + .AddMqttClient() + .BuildServiceProvider(); - await mqttClient.ConnectAsync(tcpOptions); + var factory = new MqttFactory(services); + var client = factory.CreateMqttClient(); - // ---------------------------------- - - var secureTcpOptions = new MqttClientTcpOptions - { - Server = "broker.hivemq.org", - ClientId = "TestClient", - TlsOptions = new MqttClientTlsOptions { - UseTls = true, - IgnoreCertificateChainErrors = true, - IgnoreCertificateRevocationErrors = true, - AllowUntrustedCertificates = true + // Create TCP based options using the builder + var options = new MqttClientOptionsBuilder() + .WithClientId("Client1") + .WithTcpServer("broker.hivemq.com") + .WithCredentials("bud", "%spencer%") + .WithTls() + .Build(); + + await client.ConnectAsync(options); } - }; - // ---------------------------------- + { + // Create TCP based options manually + var options = new MqttClientOptions + { + ClientId = "Client1", + Credentials = new MqttClientCredentials + { + Username = "bud", + Password = "%spencer%" + }, + ChannelOptions = new MqttClientTcpOptions + { + Server = "broker.hivemq.org", + TlsOptions = new MqttClientTlsOptions + { + UseTls = true + } + }, + }; + } - var wsOptions = new MqttClientWebSocketOptions - { - Uri = "broker.hivemq.com:8000/mqtt", - ClientId = "TestClient" - }; + { + // Subscribe to a topic + await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build()); - await mqttClient.ConnectAsync(wsOptions); + // Unsubscribe from a topic + await client.UnsubscribeAsync("my/topic"); + + // Publish an application message + var applicationMessage = new MqttApplicationMessageBuilder() + .WithTopic("A/B/C") + .WithPayload("Hello World") + .WithAtLeastOnceQoS() + .Build(); + + await client.PublishAsync(applicationMessage); + } + } // ---------------------------------- { + var services = new ServiceCollection() + .AddMqttServer(options => + { + options.ConnectionValidator = c => + { + if (c.ClientId.Length < 10) + { + return MqttConnectReturnCode.ConnectionRefusedIdentifierRejected; + } + + if (c.Username != "mySecretUser") + { + return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; + } + + if (c.Password != "mySecretPassword") + { + return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; + } + + return MqttConnectReturnCode.ConnectionAccepted; + }; + }) + .BuildServiceProvider(); + + var factory = new MqttFactory(services); var mqttServer = factory.CreateMqttServer(); await mqttServer.StartAsync(); @@ -282,7 +306,7 @@ namespace MQTTnet.TestApp.UniversalWindows } // ---------------------------------- - + // For UWP apps: MqttTcpChannel.CustomIgnorableServerCertificateErrorsResolver = o => { From 8fa91736b64ab34e2c6f27547058e453dff4bfd7 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 25 Oct 2017 22:46:19 +0200 Subject: [PATCH 5/8] Fix unit tests --- Frameworks/MQTTnet.NetStandard/MqttFactory.cs | 4 +- .../ServiceCollectionExtensions.cs | 3 +- .../MQTTnet.UniversalWindows/MqttFactory.cs | 2 +- .../ServiceCollectionExtensions.cs | 3 +- MQTTnet.Core/Client/IMqttClientFactory.cs | 2 +- MQTTnet.Core/Server/MqttServer.cs | 1 + Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 78 ++++++------------- 7 files changed, 31 insertions(+), 62 deletions(-) diff --git a/Frameworks/MQTTnet.NetStandard/MqttFactory.cs b/Frameworks/MQTTnet.NetStandard/MqttFactory.cs index b725f034..7d88480f 100644 --- a/Frameworks/MQTTnet.NetStandard/MqttFactory.cs +++ b/Frameworks/MQTTnet.NetStandard/MqttFactory.cs @@ -100,9 +100,9 @@ namespace MQTTnet return _serviceProvider.GetRequiredService(); } - public ManagedMqttClient CreateManagedMqttClient() + public IManagedMqttClient CreateManagedMqttClient() { - return _serviceProvider.GetRequiredService(); + return _serviceProvider.GetRequiredService(); } public IMqttServer CreateMqttServer() diff --git a/Frameworks/MQTTnet.NetStandard/ServiceCollectionExtensions.cs b/Frameworks/MQTTnet.NetStandard/ServiceCollectionExtensions.cs index 019f39ba..fdecb32a 100644 --- a/Frameworks/MQTTnet.NetStandard/ServiceCollectionExtensions.cs +++ b/Frameworks/MQTTnet.NetStandard/ServiceCollectionExtensions.cs @@ -52,8 +52,9 @@ namespace MQTTnet services.AddTransient(); services.AddTransient(); - services.AddTransient(); + services.AddTransient(); services.AddTransient(); + services.AddTransient(); services.AddTransient(); return services; diff --git a/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs b/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs index 68fecd7f..24826cb7 100644 --- a/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs +++ b/Frameworks/MQTTnet.UniversalWindows/MqttFactory.cs @@ -95,7 +95,7 @@ namespace MQTTnet return _serviceProvider.GetRequiredService(); } - public ManagedMqttClient CreateManagedMqttClient() + public IManagedMqttClient CreateManagedMqttClient() { return _serviceProvider.GetRequiredService(); } diff --git a/Frameworks/MQTTnet.UniversalWindows/ServiceCollectionExtensions.cs b/Frameworks/MQTTnet.UniversalWindows/ServiceCollectionExtensions.cs index f8f647a9..9b8e6f7b 100644 --- a/Frameworks/MQTTnet.UniversalWindows/ServiceCollectionExtensions.cs +++ b/Frameworks/MQTTnet.UniversalWindows/ServiceCollectionExtensions.cs @@ -46,8 +46,9 @@ namespace MQTTnet services.AddTransient(); services.AddTransient(); - services.AddTransient(); + services.AddTransient(); services.AddTransient(); + services.AddTransient(); services.AddTransient(); return services; diff --git a/MQTTnet.Core/Client/IMqttClientFactory.cs b/MQTTnet.Core/Client/IMqttClientFactory.cs index 7547906e..c8b5ef3e 100644 --- a/MQTTnet.Core/Client/IMqttClientFactory.cs +++ b/MQTTnet.Core/Client/IMqttClientFactory.cs @@ -6,6 +6,6 @@ namespace MQTTnet.Core.Client { IMqttClient CreateMqttClient(); - ManagedMqttClient CreateManagedMqttClient(); + IManagedMqttClient CreateManagedMqttClient(); } } \ No newline at end of file diff --git a/MQTTnet.Core/Server/MqttServer.cs b/MQTTnet.Core/Server/MqttServer.cs index 0d208c09..c651e826 100644 --- a/MQTTnet.Core/Server/MqttServer.cs +++ b/MQTTnet.Core/Server/MqttServer.cs @@ -28,6 +28,7 @@ namespace MQTTnet.Core.Server { throw new ArgumentNullException(nameof(adapters)); } + _adapters = adapters.ToList(); _clientSessionsManager.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e); diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index e866ca82..67a6112f 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -5,7 +5,6 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Core.Adapter; using MQTTnet.Core.Client; -using MQTTnet.Core.Packets; using MQTTnet.Core.Protocol; using MQTTnet.Core.Server; using Microsoft.Extensions.DependencyInjection; @@ -52,12 +51,7 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_WillMessage() { var serverAdapter = new TestMqttServerAdapter(); - var services = new ServiceCollection() - .AddMqttServer() - .AddSingleton(serverAdapter) - .BuildServiceProvider(); - - var s = services.GetRequiredService(); + var s = new MqttFactory().CreateMqttServer(); await s.StartAsync(); var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build(); @@ -81,12 +75,7 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_Unsubscribe() { var serverAdapter = new TestMqttServerAdapter(); - var services = new ServiceCollection() - .AddMqttServer() - .AddSingleton(serverAdapter) - .BuildServiceProvider(); - - var s = services.GetRequiredService(); + var s = new MqttFactory().CreateMqttServer(); await s.StartAsync(); var c1 = await serverAdapter.ConnectTestClient(s, "c1"); @@ -122,12 +111,7 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_Publish() { var serverAdapter = new TestMqttServerAdapter(); - var services = new ServiceCollection() - .AddMqttServer() - .AddSingleton(serverAdapter) - .BuildServiceProvider(); - - var s = services.GetRequiredService(); + var s = new MqttFactory().CreateMqttServer(); await s.StartAsync(); var c1 = await serverAdapter.ConnectTestClient(s, "c1"); @@ -150,22 +134,17 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_NoRetainedMessage() { var serverAdapter = new TestMqttServerAdapter(); - var services = new ServiceCollection() - .AddMqttServer() - .AddSingleton(serverAdapter) - .BuildServiceProvider(); - - var s = services.GetRequiredService(); + var s = new MqttFactory().CreateMqttServer(); await s.StartAsync(); var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, false)); + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).Build()); await c1.DisconnectAsync(); var c2 = await serverAdapter.ConnectTestClient(s, "c2"); var receivedMessagesCount = 0; c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce)); + await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); await Task.Delay(500); @@ -178,22 +157,17 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_RetainedMessage() { var serverAdapter = new TestMqttServerAdapter(); - var services = new ServiceCollection() - .AddMqttServer() - .AddSingleton(serverAdapter) - .BuildServiceProvider(); - - var s = services.GetRequiredService(); + var s = new MqttFactory().CreateMqttServer(); await s.StartAsync(); var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, true)); + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); await c1.DisconnectAsync(); var c2 = await serverAdapter.ConnectTestClient(s, "c2"); var receivedMessagesCount = 0; c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce)); + await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); await Task.Delay(500); @@ -206,23 +180,18 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_ClearRetainedMessage() { var serverAdapter = new TestMqttServerAdapter(); - var services = new ServiceCollection() - .AddMqttServer() - .AddSingleton(serverAdapter) - .BuildServiceProvider(); - - var s = services.GetRequiredService(); + var s = new MqttFactory().CreateMqttServer(); await s.StartAsync(); var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, true)); - await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[0], MqttQualityOfServiceLevel.AtLeastOnce, true)); + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build()); await c1.DisconnectAsync(); var c2 = await serverAdapter.ConnectTestClient(s, "c2"); var receivedMessagesCount = 0; c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce)); + await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); await Task.Delay(500); @@ -238,15 +207,15 @@ namespace MQTTnet.Core.Tests var serverAdapter = new TestMqttServerAdapter(); var services = new ServiceCollection() - .AddMqttServer() + .AddMqttServer(options => options.Storage = storage) .AddSingleton(serverAdapter) .BuildServiceProvider(); - var s = services.GetRequiredService(); + var s = new MqttFactory().CreateMqttServer(); await s.StartAsync(); var c1 = await serverAdapter.ConnectTestClient(s, "c1"); - await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, true)); + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); await c1.DisconnectAsync(); await s.StopAsync(); @@ -257,7 +226,7 @@ namespace MQTTnet.Core.Tests var c2 = await serverAdapter.ConnectTestClient(s, "c2"); var receivedMessagesCount = 0; c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; - await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce)); + await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)); await Task.Delay(500); @@ -269,18 +238,15 @@ namespace MQTTnet.Core.Tests [TestMethod] public async Task MqttServer_InterceptMessage() { - var options = new MqttServerOptions + MqttApplicationMessage Interceptor(MqttApplicationMessage message) { - ApplicationMessageInterceptor = message => - { - message.Payload = Encoding.ASCII.GetBytes("extended"); - return message; - } - }; + message.Payload = Encoding.ASCII.GetBytes("extended"); + return message; + } var serverAdapter = new TestMqttServerAdapter(); var services = new ServiceCollection() - .AddMqttServer() + .AddMqttServer(options => options.ApplicationMessageInterceptor = Interceptor) .AddSingleton(serverAdapter) .BuildServiceProvider(); From 418f2b2c3fb4f9aeb54c409b0e077cf85b049f0a Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 25 Oct 2017 23:11:42 +0200 Subject: [PATCH 6/8] Refactoring --- ...MqttApplicationMessageReceivedEventArgs.cs | 5 ++++- MQTTnet.Core/Client/MqttClient.cs | 2 +- MQTTnet.Core/Client/MqttClientExtensions.cs | 3 +-- MQTTnet.Core/Server/IMqttServer.cs | 6 ++---- ...MqttApplicationMessageReceivedEventArgs.cs | 17 ----------------- .../Server/MqttClientSessionsManager.cs | 3 ++- MQTTnet.Core/Server/MqttServer.cs | 19 +++++++++++++++---- MQTTnet.Core/Server/MqttServerOptions.cs | 2 +- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 16 ++++++++++++---- Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs | 3 ++- 10 files changed, 40 insertions(+), 36 deletions(-) delete mode 100644 MQTTnet.Core/Server/MqttApplicationMessageReceivedEventArgs.cs diff --git a/MQTTnet.Core/Client/MqttApplicationMessageReceivedEventArgs.cs b/MQTTnet.Core/Client/MqttApplicationMessageReceivedEventArgs.cs index 9e120281..98ebce5a 100644 --- a/MQTTnet.Core/Client/MqttApplicationMessageReceivedEventArgs.cs +++ b/MQTTnet.Core/Client/MqttApplicationMessageReceivedEventArgs.cs @@ -4,11 +4,14 @@ namespace MQTTnet.Core.Client { public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs { - public MqttApplicationMessageReceivedEventArgs(MqttApplicationMessage applicationMessage) + public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage) { + ClientId = clientId; ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); } + public string ClientId { get; } + public MqttApplicationMessage ApplicationMessage { get; } } } diff --git a/MQTTnet.Core/Client/MqttClient.cs b/MQTTnet.Core/Client/MqttClient.cs index e33b2581..b54288c7 100644 --- a/MQTTnet.Core/Client/MqttClient.cs +++ b/MQTTnet.Core/Client/MqttClient.cs @@ -305,7 +305,7 @@ namespace MQTTnet.Core.Client try { var applicationMessage = publishPacket.ToApplicationMessage(); - ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage)); + ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage)); } catch (Exception exception) { diff --git a/MQTTnet.Core/Client/MqttClientExtensions.cs b/MQTTnet.Core/Client/MqttClientExtensions.cs index f664dbd8..ff61548b 100644 --- a/MQTTnet.Core/Client/MqttClientExtensions.cs +++ b/MQTTnet.Core/Client/MqttClientExtensions.cs @@ -2,13 +2,12 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; -using MQTTnet.Core.Packets; namespace MQTTnet.Core.Client { public static class MqttClientExtensions { - public static Task PublishAsync(this IMqttClient client, params MqttApplicationMessage[] applicationMessages) + public static Task PublishAsync(this IApplicationMessagePublisher client, params MqttApplicationMessage[] applicationMessages) { if (client == null) throw new ArgumentNullException(nameof(client)); if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); diff --git a/MQTTnet.Core/Server/IMqttServer.cs b/MQTTnet.Core/Server/IMqttServer.cs index db3e1b0d..58c4b74c 100644 --- a/MQTTnet.Core/Server/IMqttServer.cs +++ b/MQTTnet.Core/Server/IMqttServer.cs @@ -1,18 +1,16 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; +using MQTTnet.Core.Client; namespace MQTTnet.Core.Server { - public interface IMqttServer + public interface IMqttServer : IApplicationMessageReceiver, IApplicationMessagePublisher { event EventHandler ClientConnected; event EventHandler ClientDisconnected; - event EventHandler ApplicationMessageReceived; - IList GetConnectedClients(); - void Publish(MqttApplicationMessage applicationMessage); Task StartAsync(); Task StopAsync(); diff --git a/MQTTnet.Core/Server/MqttApplicationMessageReceivedEventArgs.cs b/MQTTnet.Core/Server/MqttApplicationMessageReceivedEventArgs.cs deleted file mode 100644 index 4f2cc801..00000000 --- a/MQTTnet.Core/Server/MqttApplicationMessageReceivedEventArgs.cs +++ /dev/null @@ -1,17 +0,0 @@ -using System; - -namespace MQTTnet.Core.Server -{ - public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs - { - public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage) - { - ClientId = clientId; - ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); - } - - public string ClientId { get; } - - public MqttApplicationMessage ApplicationMessage { get; } - } -} diff --git a/MQTTnet.Core/Server/MqttClientSessionsManager.cs b/MQTTnet.Core/Server/MqttClientSessionsManager.cs index b621b835..09e07811 100644 --- a/MQTTnet.Core/Server/MqttClientSessionsManager.cs +++ b/MQTTnet.Core/Server/MqttClientSessionsManager.cs @@ -11,6 +11,7 @@ using MQTTnet.Core.Protocol; using MQTTnet.Core.Serializer; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; +using MQTTnet.Core.Client; namespace MQTTnet.Core.Server { @@ -28,9 +29,9 @@ namespace MQTTnet.Core.Server _mqttClientSesssionFactory = mqttClientSesssionFactory ?? throw new ArgumentNullException(nameof(mqttClientSesssionFactory)); } - public event EventHandler ApplicationMessageReceived; public event EventHandler ClientConnected; public event EventHandler ClientDisconnected; + public event EventHandler ApplicationMessageReceived; public MqttClientRetainedMessagesManager RetainedMessagesManager { get; } public MqttServerOptions Options { get; } diff --git a/MQTTnet.Core/Server/MqttServer.cs b/MQTTnet.Core/Server/MqttServer.cs index c651e826..16e2e2ed 100644 --- a/MQTTnet.Core/Server/MqttServer.cs +++ b/MQTTnet.Core/Server/MqttServer.cs @@ -6,6 +6,7 @@ using MQTTnet.Core.Adapter; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using System.Linq; +using MQTTnet.Core.Client; namespace MQTTnet.Core.Server { @@ -45,12 +46,22 @@ namespace MQTTnet.Core.Server public event EventHandler ClientDisconnected; public event EventHandler ApplicationMessageReceived; - public void Publish(MqttApplicationMessage applicationMessage) + public Task PublishAsync(IEnumerable applicationMessages) { - if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); - _options.ApplicationMessageInterceptor?.Invoke(applicationMessage); - _clientSessionsManager.DispatchApplicationMessage(null, applicationMessage); + if (_cancellationTokenSource == null) + { + throw new InvalidOperationException("The server is not started."); + } + + foreach (var applicationMessage in applicationMessages) + { + _options.ApplicationMessageInterceptor?.Invoke(applicationMessage); + _clientSessionsManager.DispatchApplicationMessage(null, applicationMessage); + } + + return Task.FromResult(0); } public async Task StartAsync() diff --git a/MQTTnet.Core/Server/MqttServerOptions.cs b/MQTTnet.Core/Server/MqttServerOptions.cs index 4c6f7adc..25f7cc8e 100644 --- a/MQTTnet.Core/Server/MqttServerOptions.cs +++ b/MQTTnet.Core/Server/MqttServerOptions.cs @@ -4,7 +4,7 @@ using MQTTnet.Core.Protocol; namespace MQTTnet.Core.Server { - public sealed class MqttServerOptions + public class MqttServerOptions { public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions(); diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 67a6112f..d548ff08 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -122,7 +122,7 @@ namespace MQTTnet.Core.Tests var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce)); - s.Publish(message); + s.PublishAsync(message).Wait(); await Task.Delay(500); await s.StopAsync(); @@ -180,7 +180,13 @@ namespace MQTTnet.Core.Tests public async Task MqttServer_ClearRetainedMessage() { var serverAdapter = new TestMqttServerAdapter(); - var s = new MqttFactory().CreateMqttServer(); + var services = new ServiceCollection() + .AddLogging() + .AddMqttServer() // TODO: Is there maybe an easier way for the library user to set the options? + .AddSingleton(serverAdapter) + .BuildServiceProvider(); + + var s = new MqttFactory(services).CreateMqttServer(); await s.StartAsync(); var c1 = await serverAdapter.ConnectTestClient(s, "c1"); @@ -207,11 +213,12 @@ namespace MQTTnet.Core.Tests var serverAdapter = new TestMqttServerAdapter(); var services = new ServiceCollection() - .AddMqttServer(options => options.Storage = storage) + .AddLogging() + .AddMqttServer(options => options.Storage = storage) // TODO: Is there maybe an easier way for the library user to set the options? .AddSingleton(serverAdapter) .BuildServiceProvider(); - var s = new MqttFactory().CreateMqttServer(); + var s = new MqttFactory(services).CreateMqttServer(); // TODO: Like here? await s.StartAsync(); var c1 = await serverAdapter.ConnectTestClient(s, "c1"); @@ -246,6 +253,7 @@ namespace MQTTnet.Core.Tests var serverAdapter = new TestMqttServerAdapter(); var services = new ServiceCollection() + .AddLogging() .AddMqttServer(options => options.ApplicationMessageInterceptor = Interceptor) .AddSingleton(serverAdapter) .BuildServiceProvider(); diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs index a71978c3..c5da4e5a 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs @@ -8,6 +8,7 @@ using Microsoft.Extensions.FileProviders; using Microsoft.Extensions.Logging; using MQTTnet.AspnetCore; using MQTTnet.Core; +using MQTTnet.Core.Client; namespace MQTTnet.TestApp.AspNetCore2 { @@ -29,7 +30,7 @@ namespace MQTTnet.TestApp.AspNetCore2 while (true) { - server.Publish(msg.Build()); + server.PublishAsync(msg.Build()).Wait(); await Task.Delay(TimeSpan.FromSeconds(2)); msg.WithPayload("Mqtt is still awesome at " + DateTime.Now); } From 3b366f91fc6cb9cd218024b17dd79d05765e134b Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 25 Oct 2017 23:26:15 +0200 Subject: [PATCH 7/8] Update docs --- Build/MQTTnet.nuspec | 5 +++++ README.md | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 1ed11881..2403cc84 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -13,12 +13,17 @@ * [Core] Merged the .NET Framwork and netstandard projects (Thanks to @JanEggers) * [Core] Migrated the trace to a non-static approach (Breaking Change!) * [Core] Added a builder for application messages using a fluent API +* [Core] Introduced CI (Thanks to @JanEggers) +* [Core] Added interfaces for publishing and receiving which applies to every client and server (Thanks to @ChristianRiedl) * [Client] Added a first version of a managed client which will manage the connection, subscription etc. automatically (Thanks to @JTrotta) * [Client] The session state response from the server is now returned in the _ConnectAsync_ method and also part of the _Connected_ event args * [Client] Added a _TopicFilterBuilder_ using a fluent API (Namespace Changes!) * [Client] Added several new options for the WebSocket channel (Thanks to @ChristianRiedl) +* [Client] Refactored the options and added a builder using a fluent API (Breaking Change!) +* [Client] Added more options for WebSocket connections like RequestHeaders, SubProtocol etc. * [Server] Added support for WebSockets via ASP.NET Core 2.0 (Thanks to @ChristianRiedl) * [Server] Added support for a custom application message interceptor +* [Server] Fixed an issue with dropped connections on UWP (Thanks to @haeberle) Copyright Christian Kratky 2016-2017 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation diff --git a/README.md b/README.md index 2bd607a8..6e99af3f 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov * Performance optimized (processing ~27.000 messages / second)* * Interfaces included for mocking and testing * Access to internal trace messages -* Unit tested (67+ tests) +* Unit tested (70+ tests) \* Tested on local machine with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetFramework_. From f85721fbf50c1de320a277041088a2513d8106de Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 25 Oct 2017 23:39:57 +0200 Subject: [PATCH 8/8] Restore console log of test app --- Tests/MQTTnet.TestApp.NetCore/ClientTest.cs | 14 +++++++------- Tests/MQTTnet.TestApp.NetCore/ServerTest.cs | 3 ++- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs b/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs index cccdff66..df3cd760 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs @@ -17,14 +17,14 @@ namespace MQTTnet.TestApp.NetCore { ClientId = "XYZ", CleanSession = true, - ////ChannelOptions = new MqttClientTcpOptions - ////{ - //// Server = "localhost" - ////}, - ChannelOptions = new MqttClientWebSocketOptions + ChannelOptions = new MqttClientTcpOptions { - Uri = "localhost" - } + Server = "localhost" + }, + ////ChannelOptions = new MqttClientWebSocketOptions + ////{ + //// Uri = "localhost" + ////} }; var factory = new MqttFactory(); diff --git a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs index 0432a671..a0c64623 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs @@ -49,6 +49,7 @@ namespace MQTTnet.TestApp.NetCore }); var serviceProvider = services.BuildServiceProvider(); + serviceProvider.GetRequiredService().AddConsole(); //var certificate = new X509Certificate(@"C:\certs\test\test.cer", ""); //options.TlsEndpointOptions.Certificate = certificate.Export(X509ContentType.Cert); @@ -56,7 +57,7 @@ namespace MQTTnet.TestApp.NetCore //options.DefaultEndpointOptions.IsEnabled = true; //options.TlsEndpointOptions.IsEnabled = false; - var mqttServer = serviceProvider.GetRequiredService(); + var mqttServer = new MqttFactory(serviceProvider).CreateMqttServer(); mqttServer.ClientDisconnected += (s, e) => { Console.Write("Client disconnected event fired.");