From 03bfc5f00b598ae9c9e8332cd102b3fa8677508e Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 8 Mar 2019 23:47:51 +0100 Subject: [PATCH] Refactor event handler namings. --- README.md | 13 +++- ...licationMessageProcessedHandlerDelegate.cs | 31 +++++++++ ...pplicationMessageSkippedHandlerDelegate.cs | 31 +++++++++ .../ConnectingFailedHandlerDelegate.cs | 31 +++++++++ .../IApplicationMessageProcessedHandler.cs | 9 +++ .../IApplicationMessageSkippedHandler.cs | 9 +++ .../IConnectingFailedHandler.cs | 9 +++ .../IManagedMqttClient.cs | 12 ++-- .../IManagedMqttClientOptions.cs | 1 - ...SynchronizingSubscriptionsFailedHandler.cs | 9 +++ .../ManagedMqttClient.cs | 68 +++++++++++++------ .../ManagedMqttClientExtensions.cs | 4 +- ...gs.cs => ManagedProcessFailedEventArgs.cs} | 4 +- .../MqttFactoryExtensions.cs | 7 +- ...izingSubscriptionsFailedHandlerDelegate.cs | 31 +++++++++ .../MQTTnet.Extensions.Rpc/MqttRpcClient.cs | 19 ++++-- ...cAwareApplicationMessageReceivedHandler.cs | 31 +++++++++ Source/MQTTnet/Client/MqttClient.cs | 4 +- Source/MQTTnet/Client/MqttClientExtensions.cs | 6 +- .../IMqttApplicationMessageHandler.cs | 9 --- .../IMqttApplicationMessageReceivedHandler.cs | 9 +++ ...licationMessageReceivedHandlerDelegate.cs} | 8 +-- Source/MQTTnet/IApplicationMessageReceiver.cs | 2 +- Source/MQTTnet/IMqttFactory.cs | 11 +++ Source/MQTTnet/MqttFactory.cs | 17 +++-- Source/MQTTnet/Server/IMqttServer.cs | 7 +- .../Server/IMqttServerStartedHandler.cs | 10 +++ .../Server/IMqttServerStoppedHandler.cs | 10 +++ Source/MQTTnet/Server/MqttClientConnection.cs | 2 +- .../Server/MqttClientSessionsManager.cs | 7 +- Source/MQTTnet/Server/MqttServer.cs | 21 ++++-- .../Server/MqttServerEventDispatcher.cs | 4 +- .../MqttServerStartedHandlerDelegate.cs | 31 +++++++++ .../MqttServerStoppedHandlerDelegate.cs | 31 +++++++++ Tests/MQTTnet.Core.Tests/MQTTnet.Tests.csproj | 1 + .../MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs | 2 +- Tests/MQTTnet.Core.Tests/MqttFactoryTests.cs | 2 +- Tests/MQTTnet.Core.Tests/RPC_Tests.cs | 55 +++++++++++++++ .../MQTTnet.Core.Tests/RoundtripTimeTests.cs | 2 +- Tests/MQTTnet.Core.Tests/Server_Tests.cs | 2 +- Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs | 5 +- Tests/MQTTnet.TestApp.NetCore/ClientTest.cs | 2 +- .../ManagedClientTest.cs | 2 +- .../PublicBrokerTest.cs | 2 +- Tests/MQTTnet.TestApp.NetCore/ServerTest.cs | 2 +- 45 files changed, 489 insertions(+), 96 deletions(-) create mode 100644 Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageProcessedHandlerDelegate.cs create mode 100644 Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedHandlerDelegate.cs create mode 100644 Source/MQTTnet.Extensions.ManagedClient/ConnectingFailedHandlerDelegate.cs create mode 100644 Source/MQTTnet.Extensions.ManagedClient/IApplicationMessageProcessedHandler.cs create mode 100644 Source/MQTTnet.Extensions.ManagedClient/IApplicationMessageSkippedHandler.cs create mode 100644 Source/MQTTnet.Extensions.ManagedClient/IConnectingFailedHandler.cs create mode 100644 Source/MQTTnet.Extensions.ManagedClient/ISynchronizingSubscriptionsFailedHandler.cs rename Source/MQTTnet.Extensions.ManagedClient/{MqttManagedProcessFailedEventArgs.cs => ManagedProcessFailedEventArgs.cs} (57%) create mode 100644 Source/MQTTnet.Extensions.ManagedClient/SynchronizingSubscriptionsFailedHandlerDelegate.cs create mode 100644 Source/MQTTnet.Extensions.Rpc/RpcAwareApplicationMessageReceivedHandler.cs delete mode 100644 Source/MQTTnet/Client/Receiving/IMqttApplicationMessageHandler.cs create mode 100644 Source/MQTTnet/Client/Receiving/IMqttApplicationMessageReceivedHandler.cs rename Source/MQTTnet/Client/Receiving/{MqttApplicationMessageHandlerDelegate.cs => MqttApplicationMessageReceivedHandlerDelegate.cs} (56%) create mode 100644 Source/MQTTnet/IMqttFactory.cs create mode 100644 Source/MQTTnet/Server/IMqttServerStartedHandler.cs create mode 100644 Source/MQTTnet/Server/IMqttServerStoppedHandler.cs create mode 100644 Source/MQTTnet/Server/MqttServerStartedHandlerDelegate.cs create mode 100644 Source/MQTTnet/Server/MqttServerStoppedHandlerDelegate.cs create mode 100644 Tests/MQTTnet.Core.Tests/RPC_Tests.cs diff --git a/README.md b/README.md index b74c1fbe..410f6d8b 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,8 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov * Uniform API across all supported versions of the MQTT protocol * Interfaces included for mocking and testing * Access to internal trace messages -* Unit tested (~130 tests) +* Unit tested (~150 tests) +* No external dependencies \* Tested on local machine (Intel i7 8700K) 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.NetCore_. @@ -51,6 +52,16 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov * Validate subscriptions and deny subscribing of certain topics depending on requesting clients * Connect clients with different protocol versions at the same time. +### MQTTnet Server + +_MQTTnet Server_ is a reference implementation of a MQTT server using this library. It has the following features. +* Running portable (no installation required) +* Python scripting support for manipulating messages, validation of clients etc. +* Runs und Windows, Linux, macOS, Raspberry Pi +* Supports WebSocket and TCP (with and without TLS) connections +* Provides a HTTP based API (including Swagger endpoint) +* Extensive configuration + ## Supported frameworks * .NET Standard 1.3+ diff --git a/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageProcessedHandlerDelegate.cs b/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageProcessedHandlerDelegate.cs new file mode 100644 index 00000000..e976722e --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageProcessedHandlerDelegate.cs @@ -0,0 +1,31 @@ +using System; +using System.Threading.Tasks; + +namespace MQTTnet.Extensions.ManagedClient +{ + public class ApplicationMessageProcessedHandlerDelegate : IApplicationMessageProcessedHandler + { + private readonly Func _handler; + + public ApplicationMessageProcessedHandlerDelegate(Action handler) + { + if (handler == null) throw new ArgumentNullException(nameof(handler)); + + _handler = context => + { + handler(context); + return Task.FromResult(0); + }; + } + + public ApplicationMessageProcessedHandlerDelegate(Func handler) + { + _handler = handler ?? throw new ArgumentNullException(nameof(handler)); + } + + public Task HandleApplicationMessageProcessedAsync(ApplicationMessageProcessedEventArgs eventArgs) + { + return _handler(eventArgs); + } + } +} diff --git a/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedHandlerDelegate.cs b/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedHandlerDelegate.cs new file mode 100644 index 00000000..978f30be --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedHandlerDelegate.cs @@ -0,0 +1,31 @@ +using System; +using System.Threading.Tasks; + +namespace MQTTnet.Extensions.ManagedClient +{ + public class ApplicationMessageSkippedHandlerDelegate : IApplicationMessageSkippedHandler + { + private readonly Func _handler; + + public ApplicationMessageSkippedHandlerDelegate(Action handler) + { + if (handler == null) throw new ArgumentNullException(nameof(handler)); + + _handler = eventArgs => + { + handler(eventArgs); + return Task.FromResult(0); + }; + } + + public ApplicationMessageSkippedHandlerDelegate(Func handler) + { + _handler = handler ?? throw new ArgumentNullException(nameof(handler)); + } + + public Task HandleApplicationMessageSkippedAsync(ApplicationMessageSkippedEventArgs eventArgs) + { + return _handler(eventArgs); + } + } +} diff --git a/Source/MQTTnet.Extensions.ManagedClient/ConnectingFailedHandlerDelegate.cs b/Source/MQTTnet.Extensions.ManagedClient/ConnectingFailedHandlerDelegate.cs new file mode 100644 index 00000000..85a796b9 --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/ConnectingFailedHandlerDelegate.cs @@ -0,0 +1,31 @@ +using System; +using System.Threading.Tasks; + +namespace MQTTnet.Extensions.ManagedClient +{ + public class ConnectingFailedHandlerDelegate : IConnectingFailedHandler + { + private readonly Func _handler; + + public ConnectingFailedHandlerDelegate(Action handler) + { + if (handler == null) throw new ArgumentNullException(nameof(handler)); + + _handler = eventArgs => + { + handler(eventArgs); + return Task.FromResult(0); + }; + } + + public ConnectingFailedHandlerDelegate(Func handler) + { + _handler = handler ?? throw new ArgumentNullException(nameof(handler)); + } + + public Task HandleConnectingFailedAsync(ManagedProcessFailedEventArgs eventArgs) + { + return _handler(eventArgs); + } + } +} diff --git a/Source/MQTTnet.Extensions.ManagedClient/IApplicationMessageProcessedHandler.cs b/Source/MQTTnet.Extensions.ManagedClient/IApplicationMessageProcessedHandler.cs new file mode 100644 index 00000000..08ed197a --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/IApplicationMessageProcessedHandler.cs @@ -0,0 +1,9 @@ +using System.Threading.Tasks; + +namespace MQTTnet.Extensions.ManagedClient +{ + public interface IApplicationMessageProcessedHandler + { + Task HandleApplicationMessageProcessedAsync(ApplicationMessageProcessedEventArgs eventArgs); + } +} diff --git a/Source/MQTTnet.Extensions.ManagedClient/IApplicationMessageSkippedHandler.cs b/Source/MQTTnet.Extensions.ManagedClient/IApplicationMessageSkippedHandler.cs new file mode 100644 index 00000000..e664cd2f --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/IApplicationMessageSkippedHandler.cs @@ -0,0 +1,9 @@ +using System.Threading.Tasks; + +namespace MQTTnet.Extensions.ManagedClient +{ + public interface IApplicationMessageSkippedHandler + { + Task HandleApplicationMessageSkippedAsync(ApplicationMessageSkippedEventArgs eventArgs); + } +} diff --git a/Source/MQTTnet.Extensions.ManagedClient/IConnectingFailedHandler.cs b/Source/MQTTnet.Extensions.ManagedClient/IConnectingFailedHandler.cs new file mode 100644 index 00000000..aabdbfb0 --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/IConnectingFailedHandler.cs @@ -0,0 +1,9 @@ +using System.Threading.Tasks; + +namespace MQTTnet.Extensions.ManagedClient +{ + public interface IConnectingFailedHandler + { + Task HandleConnectingFailedAsync(ManagedProcessFailedEventArgs eventArgs); + } +} diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs index 6de908fd..9919081f 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs @@ -15,13 +15,13 @@ namespace MQTTnet.Extensions.ManagedClient IMqttClientConnectedHandler ConnectedHandler { get; set; } IMqttClientDisconnectedHandler DisconnectedHandler { get; set; } + + IApplicationMessageProcessedHandler ApplicationMessageProcessedHandler { get; set; } + IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; } + + IConnectingFailedHandler ConnectingFailedHandler { get; set; } + ISynchronizingSubscriptionsFailedHandler SynchronizingSubscriptionsFailedHandler { get; set; } - event EventHandler ApplicationMessageProcessed; - event EventHandler ApplicationMessageSkipped; - - event EventHandler ConnectingFailed; - event EventHandler SynchronizingSubscriptionsFailed; - Task StartAsync(IManagedMqttClientOptions options); Task StopAsync(); diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs index 589f7acc..a705efea 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs @@ -1,5 +1,4 @@ using System; -using MQTTnet.Client; using MQTTnet.Client.Options; using MQTTnet.Server; diff --git a/Source/MQTTnet.Extensions.ManagedClient/ISynchronizingSubscriptionsFailedHandler.cs b/Source/MQTTnet.Extensions.ManagedClient/ISynchronizingSubscriptionsFailedHandler.cs new file mode 100644 index 00000000..f76f18fb --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/ISynchronizingSubscriptionsFailedHandler.cs @@ -0,0 +1,9 @@ +using System.Threading.Tasks; + +namespace MQTTnet.Extensions.ManagedClient +{ + public interface ISynchronizingSubscriptionsFailedHandler + { + Task HandleSynchronizingSubscriptionsFailedAsync(ManagedProcessFailedEventArgs eventArgs); + } +} diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 51e8accc..7340578e 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -57,17 +57,19 @@ namespace MQTTnet.Extensions.ManagedClient set => _mqttClient.DisconnectedHandler = value; } - public IMqttApplicationMessageHandler ApplicationMessageReceivedHandler + public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get => _mqttClient.ApplicationMessageReceivedHandler; set => _mqttClient.ApplicationMessageReceivedHandler = value; } - public event EventHandler ApplicationMessageProcessed; - public event EventHandler ApplicationMessageSkipped; + public IApplicationMessageProcessedHandler ApplicationMessageProcessedHandler { get; set; } - public event EventHandler ConnectingFailed; - public event EventHandler SynchronizingSubscriptionsFailed; + public IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; } + + public IConnectingFailedHandler ConnectingFailedHandler { get; set; } + + public ISynchronizingSubscriptionsFailedHandler SynchronizingSubscriptionsFailedHandler { get; set; } public async Task StartAsync(IManagedMqttClientOptions options) { @@ -126,28 +128,45 @@ namespace MQTTnet.Extensions.ManagedClient if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); ManagedMqttApplicationMessage removedMessage = null; - lock (_messageQueue) + ApplicationMessageSkippedEventArgs applicationMessageSkippedEventArgs = null; + + try { - if (_messageQueue.Count >= Options.MaxPendingMessages) + lock (_messageQueue) { - if (Options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage) + if (_messageQueue.Count >= Options.MaxPendingMessages) { - _logger.Verbose("Skipping publish of new application message because internal queue is full."); - ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(applicationMessage)); - return; + if (Options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage) + { + _logger.Verbose("Skipping publish of new application message because internal queue is full."); + applicationMessageSkippedEventArgs = new ApplicationMessageSkippedEventArgs(applicationMessage); + return; + } + + if (Options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage) + { + removedMessage = _messageQueue.RemoveFirst(); + _logger.Verbose("Removed oldest application message from internal queue because it is full."); + applicationMessageSkippedEventArgs = new ApplicationMessageSkippedEventArgs(removedMessage); + } } - if (Options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage) + _messageQueue.Enqueue(applicationMessage); + } + } + finally + { + if (applicationMessageSkippedEventArgs != null) + { + var applicationMessageSkippedHandler = ApplicationMessageSkippedHandler; + if (applicationMessageSkippedHandler != null) { - removedMessage = _messageQueue.RemoveFirst(); - _logger.Verbose("Removed oldest application message from internal queue because it is full."); - ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(removedMessage)); + await applicationMessageSkippedHandler.HandleApplicationMessageSkippedAsync(applicationMessageSkippedEventArgs).ConfigureAwait(false); } } - _messageQueue.Enqueue(applicationMessage); } - + if (_storageManager != null) { if (removedMessage != null) @@ -352,7 +371,7 @@ namespace MQTTnet.Extensions.ManagedClient } finally { - ApplicationMessageProcessed?.Invoke(this, new ApplicationMessageProcessedEventArgs(message, transmitException)); + ApplicationMessageProcessedHandler?.HandleApplicationMessageProcessedAsync(new ApplicationMessageProcessedEventArgs(message, transmitException)).GetAwaiter().GetResult(); } } @@ -395,7 +414,11 @@ namespace MQTTnet.Extensions.ManagedClient _logger.Warning(exception, "Synchronizing subscriptions failed."); _subscriptionsNotPushed = true; - SynchronizingSubscriptionsFailed?.Invoke(this, new MqttManagedProcessFailedEventArgs(exception)); + var synchronizingSubscriptionsFailedHandler = SynchronizingSubscriptionsFailedHandler; + if (SynchronizingSubscriptionsFailedHandler != null) + { + await synchronizingSubscriptionsFailedHandler.HandleSynchronizingSubscriptionsFailedAsync(new ManagedProcessFailedEventArgs(exception)).ConfigureAwait(false); + } } } @@ -413,7 +436,12 @@ namespace MQTTnet.Extensions.ManagedClient } catch (Exception exception) { - ConnectingFailed?.Invoke(this, new MqttManagedProcessFailedEventArgs(exception)); + var connectingFailedHandler = ConnectingFailedHandler; + if (connectingFailedHandler != null) + { + await connectingFailedHandler.HandleConnectingFailedAsync(new ManagedProcessFailedEventArgs(exception)).ConfigureAwait(false); + } + return ReconnectionResult.NotConnected; } } diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs index 518042c3..a0be4b10 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs @@ -68,7 +68,7 @@ namespace MQTTnet.Extensions.ManagedClient return client; } - client.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(handler); + client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(handler); return client; } @@ -81,7 +81,7 @@ namespace MQTTnet.Extensions.ManagedClient return client; } - client.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(handler); + client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(handler); return client; } diff --git a/Source/MQTTnet.Extensions.ManagedClient/MqttManagedProcessFailedEventArgs.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedProcessFailedEventArgs.cs similarity index 57% rename from Source/MQTTnet.Extensions.ManagedClient/MqttManagedProcessFailedEventArgs.cs rename to Source/MQTTnet.Extensions.ManagedClient/ManagedProcessFailedEventArgs.cs index 68dc1608..105e02a2 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/MqttManagedProcessFailedEventArgs.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedProcessFailedEventArgs.cs @@ -2,9 +2,9 @@ namespace MQTTnet.Extensions.ManagedClient { - public class MqttManagedProcessFailedEventArgs : EventArgs + public class ManagedProcessFailedEventArgs : EventArgs { - public MqttManagedProcessFailedEventArgs(Exception exception) + public ManagedProcessFailedEventArgs(Exception exception) { Exception = exception; } diff --git a/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs b/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs index 7f400c55..37393ac2 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs @@ -1,19 +1,18 @@ using System; -using MQTTnet.Client; using MQTTnet.Diagnostics; namespace MQTTnet.Extensions.ManagedClient { public static class MqttFactoryExtensions { - public static IManagedMqttClient CreateManagedMqttClient(this IMqttClientFactory factory) + public static IManagedMqttClient CreateManagedMqttClient(this IMqttFactory factory) { if (factory == null) throw new ArgumentNullException(nameof(factory)); - return new ManagedMqttClient(factory.CreateMqttClient(), new MqttNetLogger().CreateChildLogger()); + return new ManagedMqttClient(factory.CreateMqttClient(), factory.DefaultLogger.CreateChildLogger()); } - public static IManagedMqttClient CreateManagedMqttClient(this IMqttClientFactory factory, IMqttNetLogger logger) + public static IManagedMqttClient CreateManagedMqttClient(this IMqttFactory 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.ManagedClient/SynchronizingSubscriptionsFailedHandlerDelegate.cs b/Source/MQTTnet.Extensions.ManagedClient/SynchronizingSubscriptionsFailedHandlerDelegate.cs new file mode 100644 index 00000000..72d6e4a0 --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/SynchronizingSubscriptionsFailedHandlerDelegate.cs @@ -0,0 +1,31 @@ +using System; +using System.Threading.Tasks; + +namespace MQTTnet.Extensions.ManagedClient +{ + public class SynchronizingSubscriptionsFailedHandlerDelegate : ISynchronizingSubscriptionsFailedHandler + { + private readonly Func _handler; + + public SynchronizingSubscriptionsFailedHandlerDelegate(Action handler) + { + if (handler == null) throw new ArgumentNullException(nameof(handler)); + + _handler = context => + { + handler(context); + return Task.FromResult(0); + }; + } + + public SynchronizingSubscriptionsFailedHandlerDelegate(Func handler) + { + _handler = handler ?? throw new ArgumentNullException(nameof(handler)); + } + + public Task HandleSynchronizingSubscriptionsFailedAsync(ManagedProcessFailedEventArgs eventArgs) + { + return _handler(eventArgs); + } + } +} diff --git a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs index 69d3ac84..8a279c75 100644 --- a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs +++ b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs @@ -4,7 +4,6 @@ using System.Text; using System.Threading; using System.Threading.Tasks; using MQTTnet.Client; -using MQTTnet.Client.Receiving; using MQTTnet.Exceptions; using MQTTnet.Protocol; @@ -14,12 +13,17 @@ namespace MQTTnet.Extensions.Rpc { private readonly ConcurrentDictionary> _waitingCalls = new ConcurrentDictionary>(); private readonly IMqttClient _mqttClient; + private readonly RpcAwareApplicationMessageReceivedHandler _applicationMessageReceivedHandler; public MqttRpcClient(IMqttClient mqttClient) { _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient)); - _mqttClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(HandleReceivedApplicationMessageAsync); + _applicationMessageReceivedHandler = new RpcAwareApplicationMessageReceivedHandler( + mqttClient.ApplicationMessageReceivedHandler, + HandleApplicationMessageReceivedAsync); + + _mqttClient.ApplicationMessageReceivedHandler = _applicationMessageReceivedHandler; } public Task ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel) @@ -46,6 +50,11 @@ namespace MQTTnet.Extensions.Rpc throw new ArgumentException("The method name cannot contain /, + or #."); } + if (!(_mqttClient.ApplicationMessageReceivedHandler is RpcAwareApplicationMessageReceivedHandler)) + { + throw new InvalidOperationException("The application message received handler was modified."); + } + var requestTopic = $"MQTTnet.RPC/{Guid.NewGuid():N}/{methodName}"; var responseTopic = requestTopic + "/response"; @@ -103,7 +112,7 @@ namespace MQTTnet.Extensions.Rpc } } - private Task HandleReceivedApplicationMessageAsync(MqttApplicationMessageReceivedEventArgs eventArgs) + private Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs) { if (!_waitingCalls.TryRemove(eventArgs.ApplicationMessage.Topic, out var tcs)) { @@ -122,9 +131,11 @@ namespace MQTTnet.Extensions.Rpc public void Dispose() { + _mqttClient.ApplicationMessageReceivedHandler = _applicationMessageReceivedHandler.OriginalHandler; + foreach (var tcs in _waitingCalls) { - tcs.Value.SetCanceled(); + tcs.Value.TrySetCanceled(); } _waitingCalls.Clear(); diff --git a/Source/MQTTnet.Extensions.Rpc/RpcAwareApplicationMessageReceivedHandler.cs b/Source/MQTTnet.Extensions.Rpc/RpcAwareApplicationMessageReceivedHandler.cs new file mode 100644 index 00000000..9d90fb19 --- /dev/null +++ b/Source/MQTTnet.Extensions.Rpc/RpcAwareApplicationMessageReceivedHandler.cs @@ -0,0 +1,31 @@ +using System; +using System.Threading.Tasks; +using MQTTnet.Client.Receiving; + +namespace MQTTnet.Extensions.Rpc +{ + public class RpcAwareApplicationMessageReceivedHandler : IMqttApplicationMessageReceivedHandler + { + private readonly Func _handleReceivedApplicationMessageAsync; + + public RpcAwareApplicationMessageReceivedHandler( + IMqttApplicationMessageReceivedHandler originalHandler, + Func handleReceivedApplicationMessageAsync) + { + OriginalHandler = originalHandler; + _handleReceivedApplicationMessageAsync = handleReceivedApplicationMessageAsync ?? throw new ArgumentNullException(nameof(handleReceivedApplicationMessageAsync)); + } + + public IMqttApplicationMessageReceivedHandler OriginalHandler { get; } + + public async Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs) + { + if (OriginalHandler != null) + { + await OriginalHandler.HandleApplicationMessageReceivedAsync(eventArgs).ConfigureAwait(false); + } + + await _handleReceivedApplicationMessageAsync(eventArgs).ConfigureAwait(false); + } + } +} diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 75d6836f..45b99c88 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -48,7 +48,7 @@ namespace MQTTnet.Client public IMqttClientDisconnectedHandler DisconnectedHandler { get; set; } - public IMqttApplicationMessageHandler ApplicationMessageReceivedHandler { get; set; } + public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; } public bool IsConnected { get; private set; } @@ -579,7 +579,7 @@ namespace MQTTnet.Client var handler = ApplicationMessageReceivedHandler; if (handler != null) { - return handler.HandleApplicationMessageAsync( + return handler.HandleApplicationMessageReceivedAsync( new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage)); } diff --git a/Source/MQTTnet/Client/MqttClientExtensions.cs b/Source/MQTTnet/Client/MqttClientExtensions.cs index 9814f1cb..a44eb42a 100644 --- a/Source/MQTTnet/Client/MqttClientExtensions.cs +++ b/Source/MQTTnet/Client/MqttClientExtensions.cs @@ -71,7 +71,7 @@ namespace MQTTnet.Client return client; } - client.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(handler); + client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(handler); return client; } @@ -84,12 +84,12 @@ namespace MQTTnet.Client return client; } - client.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(handler); + client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(handler); return client; } - public static IMqttClient UseApplicationMessageReceivedHandler(this IMqttClient client, IMqttApplicationMessageHandler handler) + public static IMqttClient UseApplicationMessageReceivedHandler(this IMqttClient client, IMqttApplicationMessageReceivedHandler handler) { client.ApplicationMessageReceivedHandler = handler; diff --git a/Source/MQTTnet/Client/Receiving/IMqttApplicationMessageHandler.cs b/Source/MQTTnet/Client/Receiving/IMqttApplicationMessageHandler.cs deleted file mode 100644 index 5490bb04..00000000 --- a/Source/MQTTnet/Client/Receiving/IMqttApplicationMessageHandler.cs +++ /dev/null @@ -1,9 +0,0 @@ -using System.Threading.Tasks; - -namespace MQTTnet.Client.Receiving -{ - public interface IMqttApplicationMessageHandler - { - Task HandleApplicationMessageAsync(MqttApplicationMessageReceivedEventArgs eventArgs); - } -} diff --git a/Source/MQTTnet/Client/Receiving/IMqttApplicationMessageReceivedHandler.cs b/Source/MQTTnet/Client/Receiving/IMqttApplicationMessageReceivedHandler.cs new file mode 100644 index 00000000..3ea73175 --- /dev/null +++ b/Source/MQTTnet/Client/Receiving/IMqttApplicationMessageReceivedHandler.cs @@ -0,0 +1,9 @@ +using System.Threading.Tasks; + +namespace MQTTnet.Client.Receiving +{ + public interface IMqttApplicationMessageReceivedHandler + { + Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs); + } +} diff --git a/Source/MQTTnet/Client/Receiving/MqttApplicationMessageHandlerDelegate.cs b/Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedHandlerDelegate.cs similarity index 56% rename from Source/MQTTnet/Client/Receiving/MqttApplicationMessageHandlerDelegate.cs rename to Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedHandlerDelegate.cs index 399640de..1d73aaa1 100644 --- a/Source/MQTTnet/Client/Receiving/MqttApplicationMessageHandlerDelegate.cs +++ b/Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedHandlerDelegate.cs @@ -3,11 +3,11 @@ using System.Threading.Tasks; namespace MQTTnet.Client.Receiving { - public class MqttApplicationMessageHandlerDelegate : IMqttApplicationMessageHandler + public class MqttApplicationMessageReceivedHandlerDelegate : IMqttApplicationMessageReceivedHandler { private readonly Func _handler; - public MqttApplicationMessageHandlerDelegate(Action handler) + public MqttApplicationMessageReceivedHandlerDelegate(Action handler) { if (handler == null) throw new ArgumentNullException(nameof(handler)); @@ -18,12 +18,12 @@ namespace MQTTnet.Client.Receiving }; } - public MqttApplicationMessageHandlerDelegate(Func handler) + public MqttApplicationMessageReceivedHandlerDelegate(Func handler) { _handler = handler ?? throw new ArgumentNullException(nameof(handler)); } - public Task HandleApplicationMessageAsync(MqttApplicationMessageReceivedEventArgs context) + public Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs context) { return _handler(context); } diff --git a/Source/MQTTnet/IApplicationMessageReceiver.cs b/Source/MQTTnet/IApplicationMessageReceiver.cs index f34222e1..4e474856 100644 --- a/Source/MQTTnet/IApplicationMessageReceiver.cs +++ b/Source/MQTTnet/IApplicationMessageReceiver.cs @@ -4,6 +4,6 @@ namespace MQTTnet { public interface IApplicationMessageReceiver { - IMqttApplicationMessageHandler ApplicationMessageReceivedHandler { get; set; } + IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; } } } diff --git a/Source/MQTTnet/IMqttFactory.cs b/Source/MQTTnet/IMqttFactory.cs new file mode 100644 index 00000000..487d77f4 --- /dev/null +++ b/Source/MQTTnet/IMqttFactory.cs @@ -0,0 +1,11 @@ +using MQTTnet.Client; +using MQTTnet.Diagnostics; +using MQTTnet.Server; + +namespace MQTTnet +{ + public interface IMqttFactory : IMqttClientFactory, IMqttServerFactory + { + IMqttNetLogger DefaultLogger { get; } + } +} diff --git a/Source/MQTTnet/MqttFactory.cs b/Source/MQTTnet/MqttFactory.cs index 64725e26..ab6cc760 100644 --- a/Source/MQTTnet/MqttFactory.cs +++ b/Source/MQTTnet/MqttFactory.cs @@ -8,22 +8,22 @@ using MQTTnet.Server; namespace MQTTnet { - public class MqttFactory : IMqttClientFactory, IMqttServerFactory + public class MqttFactory : IMqttFactory { - private readonly IMqttNetLogger _logger; - public MqttFactory() : this(new MqttNetLogger()) { } public MqttFactory(IMqttNetLogger logger) { - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + DefaultLogger = logger ?? throw new ArgumentNullException(nameof(logger)); } + public IMqttNetLogger DefaultLogger { get; } + public IMqttClient CreateMqttClient() { - return CreateMqttClient(new MqttNetLogger()); + return CreateMqttClient(DefaultLogger); } public IMqttClient CreateMqttClient(IMqttNetLogger logger) @@ -37,7 +37,7 @@ namespace MQTTnet { if (adapterFactory == null) throw new ArgumentNullException(nameof(adapterFactory)); - return new MqttClient(adapterFactory, new MqttNetLogger()); + return new MqttClient(adapterFactory, DefaultLogger); } public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory adapterFactory) @@ -50,8 +50,7 @@ namespace MQTTnet public IMqttServer CreateMqttServer() { - var logger = new MqttNetLogger(); - return CreateMqttServer(logger); + return CreateMqttServer(DefaultLogger); } public IMqttServer CreateMqttServer(IMqttNetLogger logger) @@ -73,7 +72,7 @@ namespace MQTTnet { if (adapters == null) throw new ArgumentNullException(nameof(adapters)); - return new MqttServer(adapters, _logger.CreateChildLogger()); + return new MqttServer(adapters, DefaultLogger.CreateChildLogger()); } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/IMqttServer.cs b/Source/MQTTnet/Server/IMqttServer.cs index 1186368a..cf1681dd 100644 --- a/Source/MQTTnet/Server/IMqttServer.cs +++ b/Source/MQTTnet/Server/IMqttServer.cs @@ -1,5 +1,4 @@ -using System; -using System.Collections.Generic; +using System.Collections.Generic; using System.Threading.Tasks; using MQTTnet.Server.Status; @@ -7,8 +6,8 @@ namespace MQTTnet.Server { public interface IMqttServer : IApplicationMessageReceiver, IApplicationMessagePublisher { - event EventHandler Started; - event EventHandler Stopped; + IMqttServerStartedHandler StartedHandler { get; set; } + IMqttServerStoppedHandler StoppedHandler { get; set; } IMqttServerClientConnectedHandler ClientConnectedHandler { get; set; } IMqttServerClientDisconnectedHandler ClientDisconnectedHandler { get; set; } diff --git a/Source/MQTTnet/Server/IMqttServerStartedHandler.cs b/Source/MQTTnet/Server/IMqttServerStartedHandler.cs new file mode 100644 index 00000000..e6155f14 --- /dev/null +++ b/Source/MQTTnet/Server/IMqttServerStartedHandler.cs @@ -0,0 +1,10 @@ +using System; +using System.Threading.Tasks; + +namespace MQTTnet.Server +{ + public interface IMqttServerStartedHandler + { + Task HandleServerStartedAsync(EventArgs eventArgs); + } +} diff --git a/Source/MQTTnet/Server/IMqttServerStoppedHandler.cs b/Source/MQTTnet/Server/IMqttServerStoppedHandler.cs new file mode 100644 index 00000000..34df3236 --- /dev/null +++ b/Source/MQTTnet/Server/IMqttServerStoppedHandler.cs @@ -0,0 +1,10 @@ +using System; +using System.Threading.Tasks; + +namespace MQTTnet.Server +{ + public interface IMqttServerStoppedHandler + { + Task HandleServerStoppedAsync(EventArgs eventArgs); + } +} diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index 0610bb9e..4981b06a 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -369,7 +369,7 @@ namespace MQTTnet.Server //// } ////}); - await Task.FromResult(0); + //await Task.FromResult(0); } private async Task SendPendingPacketsAsync() diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 423638a4..34393130 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -15,6 +15,7 @@ namespace MQTTnet.Server { private readonly BlockingCollection _messageQueue = new BlockingCollection(); + private readonly SemaphoreSlim _createConnectionGate = new SemaphoreSlim(1, 1); private readonly ConcurrentDictionary _connections = new ConcurrentDictionary(); private readonly ConcurrentDictionary _sessions = new ConcurrentDictionary(); @@ -49,8 +50,6 @@ namespace MQTTnet.Server public async Task StopAsync() { - //using (await _sessionsLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) - foreach (var connection in _connections.Values) { await connection.StopAsync().ConfigureAwait(false); @@ -133,7 +132,7 @@ namespace MQTTnet.Server { if (_connections.TryGetValue(clientId, out var connection)) { - await connection.StopAsync(); + await connection.StopAsync().ConfigureAwait(false); } if (_sessions.TryRemove(clientId, out var session)) @@ -313,8 +312,6 @@ namespace MQTTnet.Server return context; } - private readonly SemaphoreSlim _createConnectionGate = new SemaphoreSlim(1, 1); - private async Task CreateConnectionAsync(IMqttChannelAdapter channelAdapter, MqttConnectPacket connectPacket) { await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false); diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 0707dec8..c3b970c4 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -30,8 +30,9 @@ namespace MQTTnet.Server _logger = logger.CreateChildLogger(nameof(MqttServer)); } - public event EventHandler Started; - public event EventHandler Stopped; + public IMqttServerStartedHandler StartedHandler { get; set; } + + public IMqttServerStoppedHandler StoppedHandler { get; set; } public IMqttServerClientConnectedHandler ClientConnectedHandler { @@ -57,7 +58,7 @@ namespace MQTTnet.Server set => _eventDispatcher.ClientUnsubscribedTopicHandler = value; } - public IMqttApplicationMessageHandler ApplicationMessageReceivedHandler + public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get => _eventDispatcher.ApplicationMessageReceivedHandler; set => _eventDispatcher.ApplicationMessageReceivedHandler = value; @@ -128,7 +129,12 @@ namespace MQTTnet.Server } _logger.Info("Started."); - Started?.Invoke(this, EventArgs.Empty); + + var startedHandler = StartedHandler; + if (startedHandler != null) + { + await startedHandler.HandleServerStartedAsync(EventArgs.Empty).ConfigureAwait(false); + } } public async Task StopAsync() @@ -151,7 +157,6 @@ namespace MQTTnet.Server } _logger.Info("Stopped."); - Stopped?.Invoke(this, EventArgs.Empty); } finally { @@ -163,6 +168,12 @@ namespace MQTTnet.Server _clientSessionsManager?.Dispose(); _clientSessionsManager = null; } + + var stoppedHandler = StoppedHandler; + if (stoppedHandler != null) + { + await stoppedHandler.HandleServerStoppedAsync(EventArgs.Empty).ConfigureAwait(false); + } } public Task ClearRetainedMessagesAsync() diff --git a/Source/MQTTnet/Server/MqttServerEventDispatcher.cs b/Source/MQTTnet/Server/MqttServerEventDispatcher.cs index cea18e1d..c85805d8 100644 --- a/Source/MQTTnet/Server/MqttServerEventDispatcher.cs +++ b/Source/MQTTnet/Server/MqttServerEventDispatcher.cs @@ -13,7 +13,7 @@ namespace MQTTnet.Server public IMqttServerClientUnsubscribedTopicHandler ClientUnsubscribedTopicHandler { get; set; } - public IMqttApplicationMessageHandler ApplicationMessageReceivedHandler { get; set; } + public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; } public Task HandleClientConnectedAsync(string clientId) { @@ -67,7 +67,7 @@ namespace MQTTnet.Server return Task.FromResult(0); } - return handler.HandleApplicationMessageAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)); + return handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)); } } } diff --git a/Source/MQTTnet/Server/MqttServerStartedHandlerDelegate.cs b/Source/MQTTnet/Server/MqttServerStartedHandlerDelegate.cs new file mode 100644 index 00000000..8a053574 --- /dev/null +++ b/Source/MQTTnet/Server/MqttServerStartedHandlerDelegate.cs @@ -0,0 +1,31 @@ +using System; +using System.Threading.Tasks; + +namespace MQTTnet.Server +{ + public class MqttServerStartedHandlerDelegate : IMqttServerStartedHandler + { + private readonly Func _handler; + + public MqttServerStartedHandlerDelegate(Action handler) + { + if (handler == null) throw new ArgumentNullException(nameof(handler)); + + _handler = eventArgs => + { + handler(eventArgs); + return Task.FromResult(0); + }; + } + + public MqttServerStartedHandlerDelegate(Func handler) + { + _handler = handler ?? throw new ArgumentNullException(nameof(handler)); + } + + public Task HandleServerStartedAsync(EventArgs eventArgs) + { + return _handler(eventArgs); + } + } +} diff --git a/Source/MQTTnet/Server/MqttServerStoppedHandlerDelegate.cs b/Source/MQTTnet/Server/MqttServerStoppedHandlerDelegate.cs new file mode 100644 index 00000000..14e6a621 --- /dev/null +++ b/Source/MQTTnet/Server/MqttServerStoppedHandlerDelegate.cs @@ -0,0 +1,31 @@ +using System; +using System.Threading.Tasks; + +namespace MQTTnet.Server +{ + public class MqttServerStoppedHandlerDelegate : IMqttServerStoppedHandler + { + private readonly Func _handler; + + public MqttServerStoppedHandlerDelegate(Action handler) + { + if (handler == null) throw new ArgumentNullException(nameof(handler)); + + _handler = eventArgs => + { + handler(eventArgs); + return Task.FromResult(0); + }; + } + + public MqttServerStoppedHandlerDelegate(Func handler) + { + _handler = handler ?? throw new ArgumentNullException(nameof(handler)); + } + + public Task HandleServerStoppedAsync(EventArgs eventArgs) + { + return _handler(eventArgs); + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/MQTTnet.Tests.csproj b/Tests/MQTTnet.Core.Tests/MQTTnet.Tests.csproj index 176cfcde..3d8d93ea 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTnet.Tests.csproj +++ b/Tests/MQTTnet.Core.Tests/MQTTnet.Tests.csproj @@ -14,6 +14,7 @@ + diff --git a/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs b/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs index 08f0fc96..93fc2a60 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs @@ -214,7 +214,7 @@ namespace MQTTnet.Tests.MQTTv5 var receivedMessages = new List(); await client1.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").WithClientId("client1").WithProtocolVersion(MqttProtocolVersion.V500).Build()); - client1.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(e => + client1.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e => { lock (receivedMessages) { diff --git a/Tests/MQTTnet.Core.Tests/MqttFactoryTests.cs b/Tests/MQTTnet.Core.Tests/MqttFactoryTests.cs index 3bdaa05d..b1a08df4 100644 --- a/Tests/MQTTnet.Core.Tests/MqttFactoryTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttFactoryTests.cs @@ -59,7 +59,7 @@ namespace MQTTnet.Tests //wait at least connect timeout or we have some log messages var tcs = new TaskCompletionSource(); - managedClient.ConnectingFailed += (s, e) => tcs.TrySetResult(null); + managedClient.ConnectingFailedHandler = new ConnectingFailedHandlerDelegate(e => tcs.TrySetResult(null)); await Task.WhenAny(Task.Delay(managedClient.Options.ClientOptions.CommunicationTimeout), tcs.Task); } finally diff --git a/Tests/MQTTnet.Core.Tests/RPC_Tests.cs b/Tests/MQTTnet.Core.Tests/RPC_Tests.cs new file mode 100644 index 00000000..b0babfa0 --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/RPC_Tests.cs @@ -0,0 +1,55 @@ +using System; +using System.Text; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Tests.Mockups; +using MQTTnet.Client; +using MQTTnet.Client.Receiving; +using MQTTnet.Exceptions; +using MQTTnet.Extensions.Rpc; +using MQTTnet.Protocol; + +namespace MQTTnet.Tests +{ + [TestClass] + public class RPC_Tests + { + [TestMethod] + public async Task Execute_Success() + { + using (var testEnvironment = new TestEnvironment()) + { + await testEnvironment.StartServerAsync(); + var responseSender = await testEnvironment.ConnectClientAsync(); + await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping"); + + responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(async e => + { + await responseSender.PublishAsync(e.ApplicationMessage.Topic + "/response", "pong"); + }); + + var requestSender = await testEnvironment.ConnectClientAsync(); + + var rpcClient = new MqttRpcClient(requestSender); + var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", MqttQualityOfServiceLevel.AtMostOnce); + + Assert.AreEqual("pong", Encoding.UTF8.GetString(response)); + } + } + + [TestMethod] + [ExpectedException(typeof(MqttCommunicationTimedOutException))] + public async Task Execute_Timeout() + { + using (var testEnvironment = new TestEnvironment()) + { + await testEnvironment.StartServerAsync(); + + var requestSender = await testEnvironment.ConnectClientAsync(); + + var rpcClient = new MqttRpcClient(requestSender); + await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce); + } + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs b/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs index 3d88a442..4cd29cef 100644 --- a/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs +++ b/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs @@ -30,7 +30,7 @@ namespace MQTTnet.Tests TaskCompletionSource response = null; - receiverClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(args => + receiverClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(args => { response?.SetResult(args.ApplicationMessage.ConvertPayloadToString()); }); diff --git a/Tests/MQTTnet.Core.Tests/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/Server_Tests.cs index 93ad66b2..5dbde0e6 100644 --- a/Tests/MQTTnet.Core.Tests/Server_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server_Tests.cs @@ -68,7 +68,7 @@ namespace MQTTnet.Tests var clientOptions = new MqttClientOptionsBuilder().WithWillMessage(willMessage); var c1 = await testEnvironment.ConnectClientAsync(); - c1.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(c => Interlocked.Increment(ref receivedMessagesCount)); + c1.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(c => Interlocked.Increment(ref receivedMessagesCount)); await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic("#").Build()); var c2 = await testEnvironment.ConnectClientAsync(clientOptions); diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs index db352f3c..3a58acf7 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs @@ -5,7 +5,6 @@ using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Hosting; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.FileProviders; -using MQTTnet; using MQTTnet.AspNetCore; using MQTTnet.Server; @@ -36,7 +35,7 @@ namespace MQTTnet.TestApp.AspNetCore2 //app.UseMqttEndpoint(); app.UseMqttServer(server => { - server.Started += async (sender, args) => + server.StartedHandler = new MqttServerStartedHandlerDelegate(async args => { var msg = new MqttApplicationMessageBuilder() .WithPayload("Mqtt is awesome") @@ -58,7 +57,7 @@ namespace MQTTnet.TestApp.AspNetCore2 await Task.Delay(TimeSpan.FromSeconds(2)); } } - }; + }); }); app.Use((context, next) => diff --git a/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs b/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs index f85a42c9..09f85777 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ClientTest.cs @@ -28,7 +28,7 @@ namespace MQTTnet.TestApp.NetCore } }; - client.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(e => + client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e => { Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###"); Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}"); diff --git a/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs b/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs index 7498e9ef..0d4f8f9d 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs @@ -35,7 +35,7 @@ namespace MQTTnet.TestApp.NetCore try { var managedClient = new MqttFactory().CreateManagedMqttClient(); - managedClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(e => + managedClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e => { Console.WriteLine(">> RECEIVED: " + e.ApplicationMessage.Topic); }); diff --git a/Tests/MQTTnet.TestApp.NetCore/PublicBrokerTest.cs b/Tests/MQTTnet.TestApp.NetCore/PublicBrokerTest.cs index 00f80c62..7f322130 100644 --- a/Tests/MQTTnet.TestApp.NetCore/PublicBrokerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/PublicBrokerTest.cs @@ -81,7 +81,7 @@ namespace MQTTnet.TestApp.NetCore var topic = Guid.NewGuid().ToString(); MqttApplicationMessage receivedMessage = null; - client.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(e => receivedMessage = e.ApplicationMessage); + client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e => receivedMessage = e.ApplicationMessage); await client.ConnectAsync(options); await client.SubscribeAsync(topic, MqttQualityOfServiceLevel.AtLeastOnce); diff --git a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs index 73aea2d7..bd626719 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs @@ -79,7 +79,7 @@ namespace MQTTnet.TestApp.NetCore var mqttServer = new MqttFactory().CreateMqttServer(); - mqttServer.ApplicationMessageReceivedHandler = new MqttApplicationMessageHandlerDelegate(e => + mqttServer.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e => { MqttNetConsoleLogger.PrintToConsole( $"'{e.ClientId}' reported '{e.ApplicationMessage.Topic}' > '{Encoding.UTF8.GetString(e.ApplicationMessage.Payload ?? new byte[0])}'",