From 7cf454952fb716b284069baf2ffd4448946b3b44 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Tue, 5 Apr 2022 20:00:49 +0200 Subject: [PATCH] Restore interfaces. --- .../MessageDeliveryBenchmark.cs | 8 +- .../MessageProcessingBenchmark.cs | 2 +- ...rocessingMqttConnectionContextBenchmark.cs | 2 +- .../MQTTnet.Benchmarks/SubscribeBenchmark.cs | 3 +- .../UnsubscribeBenchmark.cs | 5 +- .../IManagedMqttClient.cs | 52 +++++++ .../ManagedMqttClient.cs | 9 +- .../ManagedMqttClientExtensions.cs | 6 +- .../MqttFactoryExtensions.cs | 4 +- .../MQTTnet.Extensions.Rpc/IMqttRpcClient.cs | 14 ++ .../MqttFactoryExtensions.cs | 6 +- .../MQTTnet.Extensions.Rpc/MqttRpcClient.cs | 8 +- .../MqttRpcClientExtensions.cs | 2 +- .../TopicGeneration/TopicGenerationContext.cs | 2 +- .../MQTTnet.TestApp/MessageThroughputTest.cs | 37 +++-- Source/MQTTnet.TestApp/PerformanceTest.cs | 2 +- .../Client/LowLevelMqttClient_Tests.cs | 4 +- .../Client/ManagedMqttClient_Tests.cs | 2 +- .../MQTTnet.Tests/Client/MqttClient_Tests.cs | 2 +- .../MqttPacketSerialization_V3_Tests.cs | 3 +- .../MqttPacketSerialization_V5_Tests.cs | 2 +- .../TestApplicationMessageReceivedHandler.cs | 2 +- .../MQTTnet.Tests/Mockups/TestEnvironment.cs | 21 ++- Source/MQTTnet.Tests/Server/Session_Tests.cs | 4 +- Source/MQTTnet/Client/IMqttClient.cs | 38 +++++ Source/MQTTnet/Client/MqttClient.cs | 2 +- Source/MQTTnet/Client/MqttClientExtensions.cs | 23 ++- .../LowLevelClient/ILowLevelMqttClient.cs | 24 ++++ .../LowLevelClient/LowLevelMqttClient.cs | 2 +- Source/MQTTnet/MqttFactory.cs | 135 +++++++++++------- .../Internal/MqttServerKeepAliveMonitor.cs | 4 +- 31 files changed, 293 insertions(+), 137 deletions(-) create mode 100644 Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs create mode 100644 Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs create mode 100644 Source/MQTTnet/Client/IMqttClient.cs create mode 100644 Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs diff --git a/Source/MQTTnet.Benchmarks/MessageDeliveryBenchmark.cs b/Source/MQTTnet.Benchmarks/MessageDeliveryBenchmark.cs index e904e029..5bbcfdb7 100644 --- a/Source/MQTTnet.Benchmarks/MessageDeliveryBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/MessageDeliveryBenchmark.cs @@ -37,8 +37,8 @@ namespace MQTTnet.Benchmarks CancellationTokenSource _cancellationTokenSource; MqttServer _mqttServer; - List _mqttSubscriberClients; - Dictionary _mqttPublisherClientsByPublisherName; + List _mqttSubscriberClients; + Dictionary _mqttPublisherClientsByPublisherName; Dictionary> _topicsByPublisher; Dictionary _publisherByTopic; @@ -74,7 +74,7 @@ namespace MQTTnet.Benchmarks _mqttServer.StartAsync().GetAwaiter().GetResult(); // Create publisher clients - _mqttPublisherClientsByPublisherName = new Dictionary(); + _mqttPublisherClientsByPublisherName = new Dictionary(); foreach (var pt in _topicsByPublisher) { var publisherName = pt.Key; @@ -89,7 +89,7 @@ namespace MQTTnet.Benchmarks } // Create subscriber clients - _mqttSubscriberClients = new List(); + _mqttSubscriberClients = new List(); for (var i = 0; i < NumSubscribers; i++) { var mqttSubscriberClient = factory.CreateMqttClient(); diff --git a/Source/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs b/Source/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs index 175e6a2c..7ed74a58 100644 --- a/Source/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs @@ -16,7 +16,7 @@ namespace MQTTnet.Benchmarks public class MessageProcessingBenchmark { MqttServer _mqttServer; - MqttClient _mqttClient; + IMqttClient _mqttClient; MqttApplicationMessage _message; [GlobalSetup] diff --git a/Source/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs b/Source/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs index 6b65a9a6..fe3e9979 100644 --- a/Source/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs @@ -18,7 +18,7 @@ namespace MQTTnet.Benchmarks public class MessageProcessingMqttConnectionContextBenchmark { IWebHost _host; - MqttClient _mqttClient; + IMqttClient _mqttClient; MqttApplicationMessage _message; [GlobalSetup] diff --git a/Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs b/Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs index b5acf814..0449fe62 100644 --- a/Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs @@ -11,7 +11,7 @@ namespace MQTTnet.Benchmarks public class SubscribeBenchmark { MqttServer _mqttServer; - MQTTnet.Client.MqttClient _mqttClient; + IMqttClient _mqttClient; const int NumPublishers = 1; const int NumTopicsPerPublisher = 10000; @@ -54,6 +54,7 @@ namespace MQTTnet.Benchmarks var subscribeOptions = new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(topic, Protocol.MqttQualityOfServiceLevel.AtMostOnce) .Build(); + _mqttClient.SubscribeAsync(subscribeOptions).GetAwaiter().GetResult(); } } diff --git a/Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs b/Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs index 129d8d95..6b4be7e1 100644 --- a/Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs @@ -1,5 +1,4 @@ using BenchmarkDotNet.Attributes; -using BenchmarkDotNet.Jobs; using MQTTnet.Client; using MQTTnet.Server; using System.Collections.Generic; @@ -11,7 +10,7 @@ namespace MQTTnet.Benchmarks public class UnsubscribeBenchmark { MqttServer _mqttServer; - MQTTnet.Client.MqttClient _mqttClient; + IMqttClient _mqttClient; const int NumPublishers = 1; const int NumTopicsPerPublisher = 10000; @@ -40,7 +39,7 @@ namespace MQTTnet.Benchmarks foreach (var topic in _topics) { var subscribeOptions = new MqttClientSubscribeOptionsBuilder() - .WithTopicFilter(topic, Protocol.MqttQualityOfServiceLevel.AtMostOnce) + .WithTopicFilter(topic) .Build(); _mqttClient.SubscribeAsync(subscribeOptions).GetAwaiter().GetResult(); } diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs new file mode 100644 index 00000000..80ce3e38 --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs @@ -0,0 +1,52 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using MQTTnet.Client; +using MQTTnet.Packets; + +namespace MQTTnet.Extensions.ManagedClient +{ + public interface IManagedMqttClient : IDisposable + { + event Func ApplicationMessageProcessedAsync; + + event Func ApplicationMessageReceivedAsync; + + event Func ConnectedAsync; + + event Func ConnectingFailedAsync; + + event Func ConnectionStateChangedAsync; + + event Func DisconnectedAsync; + + IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; } + + IMqttClient InternalClient { get; } + + bool IsConnected { get; } + + bool IsStarted { get; } + + ManagedMqttClientOptions Options { get; } + + int PendingApplicationMessagesCount { get; } + + ISynchronizingSubscriptionsFailedHandler SynchronizingSubscriptionsFailedHandler { get; set; } + + Task EnqueueAsync(MqttApplicationMessage applicationMessage); + + Task EnqueueAsync(ManagedMqttApplicationMessage applicationMessage); + + Task PingAsync(CancellationToken cancellationToken = default); + + Task StartAsync(ManagedMqttClientOptions options); + + Task StopAsync(); + + Task SubscribeAsync(ICollection topicFilters); + + Task UnsubscribeAsync(ICollection topics); + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 4d3fed22..19c99a55 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -14,11 +14,10 @@ using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server; -using MqttClient = MQTTnet.Client.MqttClient; namespace MQTTnet.Extensions.ManagedClient { - public sealed class ManagedMqttClient : Disposable + public sealed class ManagedMqttClient : Disposable, IManagedMqttClient { readonly AsyncEvent _applicationMessageProcessedEvent = new AsyncEvent(); readonly AsyncEvent _connectingFailedEvent = new AsyncEvent(); @@ -62,7 +61,7 @@ namespace MQTTnet.Extensions.ManagedClient ManagedMqttClientStorageManager _storageManager; - public ManagedMqttClient(MqttClient mqttClient, IMqttNetLogger logger) + public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger) { InternalClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient)); @@ -112,7 +111,7 @@ namespace MQTTnet.Extensions.ManagedClient public IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; } - public MqttClient InternalClient { get; } + public IMqttClient InternalClient { get; } public bool IsConnected => InternalClient.IsConnected; @@ -203,7 +202,7 @@ namespace MQTTnet.Extensions.ManagedClient } } - public Task PingAsync(CancellationToken cancellationToken) + public Task PingAsync(CancellationToken cancellationToken = default) { return InternalClient.PingAsync(cancellationToken); } diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs index d14538cb..54dd39e7 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs @@ -13,7 +13,7 @@ namespace MQTTnet.Extensions.ManagedClient public static class ManagedMqttClientExtensions { public static Task EnqueueAsync( - this ManagedMqttClient managedMqttClient, + this IManagedMqttClient managedMqttClient, string topic, string payload = null, MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce, @@ -39,7 +39,7 @@ namespace MQTTnet.Extensions.ManagedClient } public static Task SubscribeAsync( - this ManagedMqttClient managedMqttClient, + this IManagedMqttClient managedMqttClient, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce) { @@ -60,7 +60,7 @@ namespace MQTTnet.Extensions.ManagedClient }); } - public static Task UnsubscribeAsync(this ManagedMqttClient managedMqttClient, string topic) + public static Task UnsubscribeAsync(this IManagedMqttClient managedMqttClient, string topic) { if (managedMqttClient == null) { diff --git a/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs b/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs index be9ae2a8..3349ca5d 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs @@ -10,7 +10,7 @@ namespace MQTTnet.Extensions.ManagedClient { public static class MqttFactoryExtensions { - public static ManagedMqttClient CreateManagedMqttClient(this MqttFactory factory, MqttClient mqttClient = null) + public static IManagedMqttClient CreateManagedMqttClient(this MqttFactory factory, IMqttClient mqttClient = null) { if (factory == null) throw new ArgumentNullException(nameof(factory)); @@ -22,7 +22,7 @@ namespace MQTTnet.Extensions.ManagedClient return new ManagedMqttClient(mqttClient, factory.DefaultLogger); } - public static ManagedMqttClient CreateManagedMqttClient(this MqttFactory factory, IMqttNetLogger logger) + public static IManagedMqttClient CreateManagedMqttClient(this MqttFactory factory, IMqttNetLogger logger) { if (factory == null) throw new ArgumentNullException(nameof(factory)); if (logger == null) throw new ArgumentNullException(nameof(logger)); diff --git a/Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs new file mode 100644 index 00000000..8a252899 --- /dev/null +++ b/Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs @@ -0,0 +1,14 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using MQTTnet.Protocol; + +namespace MQTTnet.Extensions.Rpc +{ + public interface IMqttRpcClient : IDisposable + { + Task ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel); + + Task ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken = default); + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs b/Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs index 42028cc0..3d4f5170 100644 --- a/Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs +++ b/Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs @@ -9,7 +9,7 @@ namespace MQTTnet.Extensions.Rpc { public static class MqttFactoryExtensions { - public static MqttRpcClient CreateMqttRpcClient(this MqttFactory factory, MqttClient mqttClient) + public static IMqttRpcClient CreateMqttRpcClient(this MqttFactory factory, IMqttClient mqttClient) { return factory.CreateMqttRpcClient(mqttClient, new MqttRpcClientOptions { @@ -17,10 +17,8 @@ namespace MQTTnet.Extensions.Rpc }); } - public static MqttRpcClient CreateMqttRpcClient(this MqttFactory factory, MqttClient mqttClient, MqttRpcClientOptions rpcClientOptions) + public static IMqttRpcClient CreateMqttRpcClient(this MqttFactory _, IMqttClient mqttClient, MqttRpcClientOptions rpcClientOptions) { - if (factory == null) throw new ArgumentNullException(nameof(factory)); - if (mqttClient == null) { throw new ArgumentNullException(nameof(mqttClient)); diff --git a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs index cf0a6f7a..29a9d6f6 100644 --- a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs +++ b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs @@ -13,13 +13,13 @@ using MQTTnet.Implementations; namespace MQTTnet.Extensions.Rpc { - public sealed class MqttRpcClient : IDisposable + public sealed class MqttRpcClient : IMqttRpcClient { readonly ConcurrentDictionary> _waitingCalls = new ConcurrentDictionary>(); - readonly MqttClient _mqttClient; + readonly IMqttClient _mqttClient; readonly MqttRpcClientOptions _options; - public MqttRpcClient(MqttClient mqttClient, MqttRpcClientOptions options) + public MqttRpcClient(IMqttClient mqttClient, MqttRpcClientOptions options) { _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient)); _options = options ?? throw new ArgumentNullException(nameof(options)); @@ -47,7 +47,7 @@ namespace MQTTnet.Extensions.Rpc } } - public async Task ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken) + public async Task ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken = default) { if (methodName == null) throw new ArgumentNullException(nameof(methodName)); diff --git a/Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs b/Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs index 86fd1ef7..8fa10468 100644 --- a/Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs +++ b/Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs @@ -11,7 +11,7 @@ namespace MQTTnet.Extensions.Rpc { public static class MqttRpcClientExtensions { - public static Task ExecuteAsync(this MqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel) + public static Task ExecuteAsync(this IMqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel) { if (client == null) throw new ArgumentNullException(nameof(client)); diff --git a/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs b/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs index fe2001c8..7185c1ba 100644 --- a/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs +++ b/Source/MQTTnet.Extensions.Rpc/Options/TopicGeneration/TopicGenerationContext.cs @@ -21,7 +21,7 @@ namespace MQTTnet.Extensions.Rpc /// public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; } - public MqttClient MqttClient { get; set; } + public IMqttClient MqttClient { get; set; } public MqttRpcClientOptions Options { get; set; } } diff --git a/Source/MQTTnet.TestApp/MessageThroughputTest.cs b/Source/MQTTnet.TestApp/MessageThroughputTest.cs index 311eaea6..67232f10 100644 --- a/Source/MQTTnet.TestApp/MessageThroughputTest.cs +++ b/Source/MQTTnet.TestApp/MessageThroughputTest.cs @@ -33,8 +33,8 @@ namespace MQTTnet.TestApp CancellationTokenSource _cancellationTokenSource; MqttServer _mqttServer; - Dictionary _mqttPublisherClientsByPublisherName; - List _mqttSubscriberClients; + Dictionary _mqttPublisherClientsByPublisherName; + List _mqttSubscriberClients; Dictionary> _topicsByPublisher; Dictionary> _singleWildcardTopicsByPublisher; @@ -85,7 +85,7 @@ namespace MQTTnet.TestApp var stopWatch = new System.Diagnostics.Stopwatch(); stopWatch.Start(); - _mqttPublisherClientsByPublisherName = new Dictionary(); + _mqttPublisherClientsByPublisherName = new Dictionary(); foreach (var pt in _topicsByPublisher) { var publisherName = pt.Key; @@ -99,10 +99,10 @@ namespace MQTTnet.TestApp } stopWatch.Stop(); - Console.Write(string.Format("{0} publisher(s) connected in {1:0.000} seconds, ", NumPublishers, stopWatch.ElapsedMilliseconds / 1000.0)); - Console.WriteLine(string.Format("connections per second: {0:0.000}", NumPublishers / (stopWatch.ElapsedMilliseconds / 1000.0))); + Console.Write($"{NumPublishers} publisher(s) connected in {stopWatch.ElapsedMilliseconds / 1000.0:0.000} seconds, "); + Console.WriteLine($"connections per second: {NumPublishers / (stopWatch.ElapsedMilliseconds / 1000.0):0.000}"); - _mqttSubscriberClients = new List(); + _mqttSubscriberClients = new List(); for (var i = 0; i < NumSubscribers; ++i) { var mqttClient = factory.CreateMqttClient(); @@ -147,7 +147,6 @@ namespace MQTTnet.TestApp _mqttServer.Dispose(); } - /// /// Measure no-wildcard topic subscription message exchange performance /// @@ -174,16 +173,16 @@ namespace MQTTnet.TestApp /// - /// Subcribe to all topics, then run message exchange + /// Subscribe to all topics, then run message exchange /// public async Task ProcessMessages(Dictionary> topicsByPublisher, string topicTypeDescription) { var numTopics = CountTopics(topicsByPublisher); Console.WriteLine(); - Console.Write(string.Format("Subscribing to {0} topics ", numTopics)); - ConsoleWriteInfo(string.Format("({0})", topicTypeDescription)); - Console.WriteLine(string.Format(" for {0} subscriber(s)...", NumSubscribers)); + Console.Write($"Subscribing to {numTopics} topics "); + ConsoleWriteInfo($"({topicTypeDescription})"); + Console.WriteLine($" for {NumSubscribers} subscriber(s)..."); var stopWatch = new System.Diagnostics.Stopwatch(); stopWatch.Start(); @@ -204,12 +203,12 @@ namespace MQTTnet.TestApp stopWatch.Stop(); - Console.Write(string.Format("{0} subscriber(s) subscribed in {1:0.000} seconds, ", NumSubscribers, stopWatch.ElapsedMilliseconds / 1000.0)); - Console.WriteLine(string.Format("subscribe calls per second: {0:0.000}", numTopics / (stopWatch.ElapsedMilliseconds / 1000.0))); + Console.Write($"{NumSubscribers} subscriber(s) subscribed in {stopWatch.ElapsedMilliseconds / 1000.0:0.000} seconds, "); + Console.WriteLine($"subscribe calls per second: {numTopics / (stopWatch.ElapsedMilliseconds / 1000.0):0.000}"); await PublishAllAsync(); - Console.WriteLine(string.Format("Unsubscribing {0} topics ({1}) for {2} subscriber(s)...", numTopics, topicTypeDescription, NumSubscribers)); + Console.WriteLine($"Unsubscribing {numTopics} topics ({topicTypeDescription}) for {NumSubscribers} subscriber(s)..."); stopWatch.Restart(); @@ -234,11 +233,10 @@ namespace MQTTnet.TestApp stopWatch.Stop(); - Console.Write(string.Format("{0} subscriber(s) unsubscribed in {1:0.000} seconds, ", NumSubscribers, stopWatch.ElapsedMilliseconds / 1000.0)); - Console.WriteLine(string.Format("unsubscribe calls per second: {0:0.000}", numTopics / (stopWatch.ElapsedMilliseconds / 1000.0))); + Console.Write($"{NumSubscribers} subscriber(s) unsubscribed in {stopWatch.ElapsedMilliseconds / 1000.0:0.000} seconds, "); + Console.WriteLine($"unsubscribe calls per second: {numTopics / (stopWatch.ElapsedMilliseconds / 1000.0):0.000}"); } - /// /// Publish messages in batches of NumPublishCallsPerBatch, wait for messages sent to subscriber /// @@ -289,7 +287,6 @@ namespace MQTTnet.TestApp } catch { - } _cancellationTokenSource.Dispose(); @@ -305,8 +302,8 @@ namespace MQTTnet.TestApp stopWatch.Stop(); - System.Console.Write(string.Format("{0} messages published and received in {1:0.000} seconds, ", totalNumMessagesReceived, stopWatch.ElapsedMilliseconds / 1000.0)); - ConsoleWriteLineSuccess(string.Format("messages per second: {0}", (int)(totalNumMessagesReceived / (stopWatch.ElapsedMilliseconds / 1000.0)))); + Console.Write($"{totalNumMessagesReceived} messages published and received in {stopWatch.ElapsedMilliseconds / 1000.0:0.000} seconds, "); + ConsoleWriteLineSuccess($"messages per second: {(int)(totalNumMessagesReceived / (stopWatch.ElapsedMilliseconds / 1000.0))}"); } int CountTopics(Dictionary> topicsByPublisher) diff --git a/Source/MQTTnet.TestApp/PerformanceTest.cs b/Source/MQTTnet.TestApp/PerformanceTest.cs index e4189b27..03b01d05 100644 --- a/Source/MQTTnet.TestApp/PerformanceTest.cs +++ b/Source/MQTTnet.TestApp/PerformanceTest.cs @@ -201,7 +201,7 @@ namespace MQTTnet.TestApp }; } - static Task PublishSingleMessage(MqttClient client, MqttApplicationMessage applicationMessage, ref int count) + static Task PublishSingleMessage(IMqttClient client, MqttApplicationMessage applicationMessage, ref int count) { Interlocked.Increment(ref count); return Task.Run(() => client.PublishAsync(applicationMessage)); diff --git a/Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs b/Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs index 6534bf3f..ae8fcebe 100644 --- a/Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Client/LowLevelMqttClient_Tests.cs @@ -177,7 +177,7 @@ namespace MQTTnet.Tests.Client } } - async Task Authenticate(LowLevelMqttClient client) + async Task Authenticate(ILowLevelMqttClient client) { await client.SendAsync( new MqttConnectPacket @@ -193,7 +193,7 @@ namespace MQTTnet.Tests.Client return await client.ReceiveAsync(CancellationToken.None).ConfigureAwait(false) as MqttConnAckPacket; } - async Task Subscribe(LowLevelMqttClient client, string topic) + async Task Subscribe(ILowLevelMqttClient client, string topic) { await client.SendAsync( new MqttSubscribePacket diff --git a/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs b/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs index e64931ac..a590027d 100644 --- a/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs @@ -389,7 +389,7 @@ namespace MQTTnet.Tests.Client async Task CreateManagedClientAsync( TestEnvironment testEnvironment, - MqttClient underlyingClient = null, + IMqttClient underlyingClient = null, TimeSpan? connectionCheckInterval = null, string host = "localhost") { diff --git a/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs b/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs index e6dbcdc4..af7c5719 100644 --- a/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs @@ -226,7 +226,7 @@ namespace MQTTnet.Tests.Client { await testEnvironment.StartServer(); - var clients = new List(); + var clients = new List(); for (var i = 0; i < 100; i++) { clients.Add(await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a"))); diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs index 7532c34f..5ae86cc1 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs @@ -204,7 +204,7 @@ namespace MQTTnet.Tests { var disconnectPacket = new MqttDisconnectPacket { - ReasonCode = MqttDisconnectReasonCode.NormalDisconnection, + ReasonCode = MqttDisconnectReasonCode.NormalDisconnection, // MQTTv3 has no other values than this. ReasonString = "ReasonString", ServerReference = "ServerReference", SessionExpiryInterval = 234, @@ -216,7 +216,6 @@ namespace MQTTnet.Tests var deserialized = MqttPacketSerializationHelper.EncodeAndDecodePacket(disconnectPacket, MqttProtocolVersion.V311); - Assert.AreEqual(disconnectPacket.ReasonCode, deserialized.ReasonCode); Assert.AreEqual(null, deserialized.ReasonString); // Not supported in v3.1.1 Assert.AreEqual(null, deserialized.ServerReference); // Not supported in v3.1.1 diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs index bbbf1916..1d9e9740 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs @@ -166,7 +166,7 @@ namespace MQTTnet.Tests { var disconnectPacket = new MqttDisconnectPacket { - ReasonCode = MqttDisconnectReasonCode.NormalDisconnection, + ReasonCode = MqttDisconnectReasonCode.QuotaExceeded, ReasonString = "ReasonString", ServerReference = "ServerReference", SessionExpiryInterval = 234, diff --git a/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs b/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs index 0a7af12e..81bca57b 100644 --- a/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs +++ b/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs @@ -15,7 +15,7 @@ namespace MQTTnet.Tests.Mockups { readonly List _receivedEventArgs = new List(); - public TestApplicationMessageReceivedHandler(MqttClient mqttClient) + public TestApplicationMessageReceivedHandler(IMqttClient mqttClient) { mqttClient.ApplicationMessageReceivedAsync += MqttClientOnApplicationMessageReceivedAsync; } diff --git a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs index 88a3afc2..c3da9717 100644 --- a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs +++ b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs @@ -17,17 +17,16 @@ using MQTTnet.Implementations; using MQTTnet.LowLevelClient; using MQTTnet.Protocol; using MQTTnet.Server; -using MqttClient = MQTTnet.Client.MqttClient; namespace MQTTnet.Tests.Mockups { public sealed class TestEnvironment : IDisposable { readonly List _clientErrors = new List(); - readonly List _clients = new List(); + readonly List _clients = new List(); readonly List _exceptions = new List(); - readonly List _lowLevelClients = new List(); + readonly List _lowLevelClients = new List(); readonly MqttProtocolVersion _protocolVersion; readonly List _serverErrors = new List(); @@ -91,12 +90,12 @@ namespace MQTTnet.Tests.Mockups public TestContext TestContext { get; } - public Task ConnectClient() + public Task ConnectClient() { return ConnectClient(Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion)); } - public async Task ConnectClient(Action configureOptions, TimeSpan timeout = default) + public async Task ConnectClient(Action configureOptions, TimeSpan timeout = default) { if (configureOptions == null) { @@ -128,7 +127,7 @@ namespace MQTTnet.Tests.Mockups return client; } - public async Task ConnectClient(MqttClientOptionsBuilder options, TimeSpan timeout = default) + public async Task ConnectClient(MqttClientOptionsBuilder options, TimeSpan timeout = default) { if (options == null) { @@ -154,7 +153,7 @@ namespace MQTTnet.Tests.Mockups return client; } - public async Task ConnectClient(MqttClientOptions options, TimeSpan timeout = default) + public async Task ConnectClient(MqttClientOptions options, TimeSpan timeout = default) { if (options == null) { @@ -178,7 +177,7 @@ namespace MQTTnet.Tests.Mockups return client; } - public async Task ConnectLowLevelClient(Action optionsBuilder = null) + public async Task ConnectLowLevelClient(Action optionsBuilder = null) { var options = new MqttClientOptionsBuilder(); options = options.WithTcpServer("127.0.0.1", ServerPort); @@ -195,12 +194,12 @@ namespace MQTTnet.Tests.Mockups return new MqttRpcClient(await ConnectClient(), options); } - public TestApplicationMessageReceivedHandler CreateApplicationMessageHandler(MqttClient mqttClient) + public TestApplicationMessageReceivedHandler CreateApplicationMessageHandler(IMqttClient mqttClient) { return new TestApplicationMessageReceivedHandler(mqttClient); } - public MqttClient CreateClient() + public IMqttClient CreateClient() { lock (_clients) { @@ -228,7 +227,7 @@ namespace MQTTnet.Tests.Mockups } } - public LowLevelMqttClient CreateLowLevelClient() + public ILowLevelMqttClient CreateLowLevelClient() { lock (_clients) { diff --git a/Source/MQTTnet.Tests/Server/Session_Tests.cs b/Source/MQTTnet.Tests/Server/Session_Tests.cs index c11a8780..b4b9a8ab 100644 --- a/Source/MQTTnet.Tests/Server/Session_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Session_Tests.cs @@ -277,7 +277,7 @@ namespace MQTTnet.Tests.Server } } - async Task ConnectAndSubscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive) + async Task ConnectAndSubscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive) { try { @@ -302,7 +302,7 @@ namespace MQTTnet.Tests.Server } } - async Task TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options) + async Task TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options) { try { diff --git a/Source/MQTTnet/Client/IMqttClient.cs b/Source/MQTTnet/Client/IMqttClient.cs new file mode 100644 index 00000000..a9ded3a9 --- /dev/null +++ b/Source/MQTTnet/Client/IMqttClient.cs @@ -0,0 +1,38 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using MQTTnet.Diagnostics; + +namespace MQTTnet.Client +{ + public interface IMqttClient : IDisposable + { + event Func ConnectedAsync; + + event Func ConnectingAsync; + + event Func DisconnectedAsync; + + event Func ApplicationMessageReceivedAsync; + + event Func InspectPackage; + + bool IsConnected { get; } + + MqttClientOptions Options { get; } + + Task ConnectAsync(MqttClientOptions options, CancellationToken cancellationToken = default); + + Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken = default); + + Task PingAsync(CancellationToken cancellationToken = default); + + Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default); + + Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default); + + Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default); + + Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index d231d870..9be1b725 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -17,7 +17,7 @@ using MQTTnet.Implementations; namespace MQTTnet.Client { - public sealed class MqttClient : Disposable + public sealed class MqttClient : Disposable, IMqttClient { readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); diff --git a/Source/MQTTnet/Client/MqttClientExtensions.cs b/Source/MQTTnet/Client/MqttClientExtensions.cs index b48dac5a..8e8ffb65 100644 --- a/Source/MQTTnet/Client/MqttClientExtensions.cs +++ b/Source/MQTTnet/Client/MqttClientExtensions.cs @@ -14,7 +14,7 @@ namespace MQTTnet.Client { public static class MqttClientExtensions { - public static Task DisconnectAsync(this MqttClient client, MqttClientDisconnectReason reason = MqttClientDisconnectReason.NormalDisconnection, string reasonString = null) + public static Task DisconnectAsync(this IMqttClient client, MqttClientDisconnectReason reason = MqttClientDisconnectReason.NormalDisconnection, string reasonString = null) { if (client == null) { @@ -31,7 +31,7 @@ namespace MQTTnet.Client } public static Task PublishBinaryAsync( - this MqttClient mqttClient, + this IMqttClient mqttClient, string topic, IEnumerable payload = null, MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce, @@ -58,7 +58,7 @@ namespace MQTTnet.Client } public static Task PublishStringAsync( - this MqttClient mqttClient, + this IMqttClient mqttClient, string topic, string payload = null, MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce, @@ -69,7 +69,7 @@ namespace MQTTnet.Client return mqttClient.PublishBinaryAsync(topic, payloadBuffer, qualityOfServiceLevel, retain, cancellationToken); } - public static Task SendExtendedAuthenticationExchangeDataAsync(this MqttClient client, MqttExtendedAuthenticationExchangeData data) + public static Task SendExtendedAuthenticationExchangeDataAsync(this IMqttClient client, MqttExtendedAuthenticationExchangeData data) { if (client == null) { @@ -79,7 +79,7 @@ namespace MQTTnet.Client return client.SendExtendedAuthenticationExchangeDataAsync(data, CancellationToken.None); } - public static Task SubscribeAsync(this MqttClient mqttClient, MqttTopicFilter topicFilter, CancellationToken cancellationToken = default) + public static Task SubscribeAsync(this IMqttClient mqttClient, MqttTopicFilter topicFilter, CancellationToken cancellationToken = default) { if (mqttClient == null) { @@ -91,14 +91,13 @@ namespace MQTTnet.Client throw new ArgumentNullException(nameof(topicFilter)); } - var subscribeOptions = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topicFilter) - .Build(); + var subscribeOptions = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topicFilter).Build(); return mqttClient.SubscribeAsync(subscribeOptions, cancellationToken); } public static Task SubscribeAsync( - this MqttClient mqttClient, + this IMqttClient mqttClient, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce, CancellationToken cancellationToken = default) @@ -113,13 +112,12 @@ namespace MQTTnet.Client throw new ArgumentNullException(nameof(topic)); } - var subscribeOptions = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topic, qualityOfServiceLevel) - .Build(); + var subscribeOptions = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topic, qualityOfServiceLevel).Build(); return mqttClient.SubscribeAsync(subscribeOptions, cancellationToken); } - public static Task UnsubscribeAsync(this MqttClient mqttClient, string topic, CancellationToken cancellationToken = default) + public static Task UnsubscribeAsync(this IMqttClient mqttClient, string topic, CancellationToken cancellationToken = default) { if (mqttClient == null) { @@ -131,8 +129,7 @@ namespace MQTTnet.Client throw new ArgumentNullException(nameof(topic)); } - var unsubscribeOptions = new MqttClientUnsubscribeOptionsBuilder().WithTopicFilter(topic) - .Build(); + var unsubscribeOptions = new MqttClientUnsubscribeOptionsBuilder().WithTopicFilter(topic).Build(); return mqttClient.UnsubscribeAsync(unsubscribeOptions, cancellationToken); } diff --git a/Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs b/Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs new file mode 100644 index 00000000..40132f28 --- /dev/null +++ b/Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs @@ -0,0 +1,24 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using MQTTnet.Client; +using MQTTnet.Diagnostics; +using MQTTnet.Packets; + +namespace MQTTnet.LowLevelClient +{ + public interface ILowLevelMqttClient : IDisposable + { + event Func InspectPackage; + + bool IsConnected { get; } + + Task ConnectAsync(MqttClientOptions options, CancellationToken cancellationToken = default); + + Task DisconnectAsync(CancellationToken cancellationToken = default); + + Task ReceiveAsync(CancellationToken cancellationToken = default); + + Task SendAsync(MqttPacket packet, CancellationToken cancellationToken = default); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs b/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs index 3e325c2e..a70b58da 100644 --- a/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs +++ b/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs @@ -14,7 +14,7 @@ using MQTTnet.Packets; namespace MQTTnet.LowLevelClient { - public sealed class LowLevelMqttClient : IDisposable + public sealed class LowLevelMqttClient : ILowLevelMqttClient { readonly IMqttClientAdapterFactory _clientAdapterFactory; readonly AsyncEvent _inspectPacketEvent = new AsyncEvent(); diff --git a/Source/MQTTnet/MqttFactory.cs b/Source/MQTTnet/MqttFactory.cs index 17ab896d..09e6bd85 100644 --- a/Source/MQTTnet/MqttFactory.cs +++ b/Source/MQTTnet/MqttFactory.cs @@ -2,15 +2,15 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. -using MQTTnet.Adapter; -using MQTTnet.Client; -using MQTTnet.Implementations; -using MQTTnet.LowLevelClient; -using MQTTnet.Server; using System; using System.Collections.Generic; using System.Linq; +using MQTTnet.Adapter; +using MQTTnet.Client; using MQTTnet.Diagnostics; +using MQTTnet.Implementations; +using MQTTnet.LowLevelClient; +using MQTTnet.Server; using MqttClient = MQTTnet.Client.MqttClient; namespace MQTTnet @@ -26,7 +26,7 @@ namespace MQTTnet public MqttFactory(IMqttNetLogger logger) { DefaultLogger = logger ?? throw new ArgumentNullException(nameof(logger)); - + _clientAdapterFactory = new MqttClientAdapterFactory(); } @@ -39,62 +39,97 @@ namespace MQTTnet public IDictionary Properties { get; } = new Dictionary(); - public MqttFactory UseClientAdapterFactory(IMqttClientAdapterFactory clientAdapterFactory) + public MqttApplicationMessageBuilder CreateApplicationMessageBuilder() { - _clientAdapterFactory = clientAdapterFactory ?? throw new ArgumentNullException(nameof(clientAdapterFactory)); - return this; + return new MqttApplicationMessageBuilder(); } - public LowLevelMqttClient CreateLowLevelMqttClient() + public MqttClientDisconnectOptionsBuilder CreateClientDisconnectOptionsBuilder() + { + return new MqttClientDisconnectOptionsBuilder(); + } + + public MqttClientOptionsBuilder CreateClientOptionsBuilder() + { + return new MqttClientOptionsBuilder(); + } + + public ILowLevelMqttClient CreateLowLevelMqttClient() { return CreateLowLevelMqttClient(DefaultLogger); } - public LowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger) + public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger) { - if (logger == null) throw new ArgumentNullException(nameof(logger)); + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } return new LowLevelMqttClient(_clientAdapterFactory, logger); } - public LowLevelMqttClient CreateLowLevelMqttClient(IMqttClientAdapterFactory clientAdapterFactory) + public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttClientAdapterFactory clientAdapterFactory) { - if (clientAdapterFactory == null) throw new ArgumentNullException(nameof(clientAdapterFactory)); + if (clientAdapterFactory == null) + { + throw new ArgumentNullException(nameof(clientAdapterFactory)); + } return new LowLevelMqttClient(_clientAdapterFactory, DefaultLogger); } - public LowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactory) + public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactory) { - if (logger == null) throw new ArgumentNullException(nameof(logger)); - if (clientAdapterFactory == null) throw new ArgumentNullException(nameof(clientAdapterFactory)); + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } + + if (clientAdapterFactory == null) + { + throw new ArgumentNullException(nameof(clientAdapterFactory)); + } return new LowLevelMqttClient(_clientAdapterFactory, logger); } - public MqttClient CreateMqttClient() + public IMqttClient CreateMqttClient() { return CreateMqttClient(DefaultLogger); } - public MqttClient CreateMqttClient(IMqttNetLogger logger) + public IMqttClient CreateMqttClient(IMqttNetLogger logger) { - if (logger == null) throw new ArgumentNullException(nameof(logger)); + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } return new MqttClient(_clientAdapterFactory, logger); } - public MqttClient CreateMqttClient(IMqttClientAdapterFactory clientAdapterFactory) + public IMqttClient CreateMqttClient(IMqttClientAdapterFactory clientAdapterFactory) { - if (clientAdapterFactory == null) throw new ArgumentNullException(nameof(clientAdapterFactory)); + if (clientAdapterFactory == null) + { + throw new ArgumentNullException(nameof(clientAdapterFactory)); + } return new MqttClient(clientAdapterFactory, DefaultLogger); } - public MqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactory) + public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactory) { - if (logger == null) throw new ArgumentNullException(nameof(logger)); - if (clientAdapterFactory == null) throw new ArgumentNullException(nameof(clientAdapterFactory)); + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } + + if (clientAdapterFactory == null) + { + throw new ArgumentNullException(nameof(clientAdapterFactory)); + } return new MqttClient(clientAdapterFactory, logger); } @@ -106,7 +141,10 @@ namespace MQTTnet public MqttServer CreateMqttServer(MqttServerOptions options, IMqttNetLogger logger) { - if (logger == null) throw new ArgumentNullException(nameof(logger)); + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } var serverAdapters = DefaultServerAdapters.Select(a => a.Invoke(this)); return CreateMqttServer(options, serverAdapters, logger); @@ -114,52 +152,53 @@ namespace MQTTnet public MqttServer CreateMqttServer(MqttServerOptions options, IEnumerable serverAdapters, IMqttNetLogger logger) { - if (serverAdapters == null) throw new ArgumentNullException(nameof(serverAdapters)); - if (logger == null) throw new ArgumentNullException(nameof(logger)); + if (serverAdapters == null) + { + throw new ArgumentNullException(nameof(serverAdapters)); + } + + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } return new MqttServer(options, serverAdapters, logger); } public MqttServer CreateMqttServer(MqttServerOptions options, IEnumerable serverAdapters) { - if (serverAdapters == null) throw new ArgumentNullException(nameof(serverAdapters)); + if (serverAdapters == null) + { + throw new ArgumentNullException(nameof(serverAdapters)); + } return new MqttServer(options, serverAdapters, DefaultLogger); } - + public MqttServerOptionsBuilder CreateServerOptionsBuilder() { return new MqttServerOptionsBuilder(); } - - public MqttClientOptionsBuilder CreateClientOptionsBuilder() - { - return new MqttClientOptionsBuilder(); - } - public MqttClientDisconnectOptionsBuilder CreateClientDisconnectOptionsBuilder() - { - return new MqttClientDisconnectOptionsBuilder(); - } - public MqttClientSubscribeOptionsBuilder CreateSubscribeOptionsBuilder() { return new MqttClientSubscribeOptionsBuilder(); } - - public MqttClientUnsubscribeOptionsBuilder CreateUnsubscribeOptionsBuilder() - { - return new MqttClientUnsubscribeOptionsBuilder(); - } - + public MqttTopicFilterBuilder CreateTopicFilterBuilder() { return new MqttTopicFilterBuilder(); } - public MqttApplicationMessageBuilder CreateApplicationMessageBuilder() + public MqttClientUnsubscribeOptionsBuilder CreateUnsubscribeOptionsBuilder() { - return new MqttApplicationMessageBuilder(); + return new MqttClientUnsubscribeOptionsBuilder(); + } + + public MqttFactory UseClientAdapterFactory(IMqttClientAdapterFactory clientAdapterFactory) + { + _clientAdapterFactory = clientAdapterFactory ?? throw new ArgumentNullException(nameof(clientAdapterFactory)); + return this; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs index 76deb057..de2bd442 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs @@ -96,10 +96,10 @@ namespace MQTTnet.Server // Values described here: [MQTT-3.1.2-24]. // If the client sends 5 sec. the server will allow up to 7.5 seconds. // If the client sends 1 sec. the server will allow up to 1.5 seconds. - var maxDurationWithoutPacket = connection.KeepAlivePeriod * 1.5D; + var maxSecondsWithoutPacket = connection.KeepAlivePeriod * 1.5D; var secondsWithoutPackage = (now - connection.Statistics.LastPacketSentTimestamp).TotalSeconds; - if (secondsWithoutPackage < maxDurationWithoutPacket) + if (secondsWithoutPackage < maxSecondsWithoutPacket) { // A packet was received before the timeout is affected. return;