From a3f6a501dc3fffe8d33e108f6fb5d11ce70a2067 Mon Sep 17 00:00:00 2001 From: Christian Date: Wed, 27 Oct 2021 23:11:20 +0200 Subject: [PATCH] Add interceptor for all packets. --- .../Extensions/ServiceCollectionExtensions.cs | 2 +- Source/MQTTnet.AspnetCore/MqttHostedServer.cs | 65 ++----------------- Source/MQTTnet/MQTTnet.csproj.DotSettings | 4 +- .../InterceptingMqttClientPublishEventArgs.cs | 0 ...rceptingMqttClientSubscriptionEventArgs.cs | 0 ...eptingMqttClientUnsubscriptionEventArgs.cs | 0 .../Events/InterceptingPacketEventArgs.cs | 30 +++++++++ ...tApplicationMessageNotConsumedEventArgs.cs | 0 .../MqttClientConnectedEventArgs.cs | 0 .../MqttClientDisconnectedEventArgs.cs | 0 .../MqttClientSubscribedTopicEventArgs.cs | 0 .../MqttClientUnsubscribedTopicEventArgs.cs | 0 .../PreparingMqttClientSessionEventArgs.cs | 0 ...ValidatingMqttClientConnectionEventArgs.cs | 0 Source/MQTTnet/Server/IMqttServer.cs | 23 ------- .../Server/Internal/MqttClientConnection.cs | 14 ++++ .../Internal/MqttServerEventContainer.cs | 4 ++ ...sageResponse.cs => MqttPublishResponse.cs} | 0 Source/MQTTnet/Server/MqttServer.cs | 14 +++- Source/MQTTnet/Server/MqttServerExtensions.cs | 8 +-- .../IMqttRetainedMessagesManager.cs | 3 +- .../IMqttServerPersistedSession.cs | 0 .../IMqttServerPersistedSessionsStorage.cs | 0 .../{ => Storage}/IMqttServerStorage.cs | 0 .../MqttTcpChannelBenchmark.cs | 2 +- 25 files changed, 76 insertions(+), 93 deletions(-) rename Source/MQTTnet/Server/{ => Events}/InterceptingMqttClientPublishEventArgs.cs (100%) rename Source/MQTTnet/Server/{ => Events}/InterceptingMqttClientSubscriptionEventArgs.cs (100%) rename Source/MQTTnet/Server/{ => Events}/InterceptingMqttClientUnsubscriptionEventArgs.cs (100%) create mode 100644 Source/MQTTnet/Server/Events/InterceptingPacketEventArgs.cs rename Source/MQTTnet/Server/{ => Events}/MqttApplicationMessageNotConsumedEventArgs.cs (100%) rename Source/MQTTnet/Server/{ => Events}/MqttClientConnectedEventArgs.cs (100%) rename Source/MQTTnet/Server/{ => Events}/MqttClientDisconnectedEventArgs.cs (100%) rename Source/MQTTnet/Server/{ => Events}/MqttClientSubscribedTopicEventArgs.cs (100%) rename Source/MQTTnet/Server/{ => Events}/MqttClientUnsubscribedTopicEventArgs.cs (100%) rename Source/MQTTnet/Server/{ => Events}/PreparingMqttClientSessionEventArgs.cs (100%) rename Source/MQTTnet/Server/{ => Events}/ValidatingMqttClientConnectionEventArgs.cs (100%) delete mode 100644 Source/MQTTnet/Server/IMqttServer.cs rename Source/MQTTnet/Server/{MqttApplicationMessageResponse.cs => MqttPublishResponse.cs} (100%) rename Source/MQTTnet/Server/{ => Storage}/IMqttRetainedMessagesManager.cs (87%) rename Source/MQTTnet/Server/{ => Storage}/IMqttServerPersistedSession.cs (100%) rename Source/MQTTnet/Server/{ => Storage}/IMqttServerPersistedSessionsStorage.cs (100%) rename Source/MQTTnet/Server/{ => Storage}/IMqttServerStorage.cs (100%) diff --git a/Source/MQTTnet.AspnetCore/Extensions/ServiceCollectionExtensions.cs b/Source/MQTTnet.AspnetCore/Extensions/ServiceCollectionExtensions.cs index 66295f5d..600fd97b 100644 --- a/Source/MQTTnet.AspnetCore/Extensions/ServiceCollectionExtensions.cs +++ b/Source/MQTTnet.AspnetCore/Extensions/ServiceCollectionExtensions.cs @@ -67,7 +67,7 @@ namespace MQTTnet.AspNetCore.Extensions services.AddSingleton(logger); services.AddSingleton(); services.AddSingleton(s => s.GetService()); - services.AddSingleton(s => s.GetService()); + services.AddSingleton(s => s.GetService()); return services; } diff --git a/Source/MQTTnet.AspnetCore/MqttHostedServer.cs b/Source/MQTTnet.AspnetCore/MqttHostedServer.cs index 10cb1078..bcffb305 100644 --- a/Source/MQTTnet.AspnetCore/MqttHostedServer.cs +++ b/Source/MQTTnet.AspnetCore/MqttHostedServer.cs @@ -3,82 +3,27 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Hosting; using MQTTnet.Adapter; -using MQTTnet.Client.Publishing; using MQTTnet.Diagnostics.Logger; using MQTTnet.Server; namespace MQTTnet.AspNetCore { - public sealed class MqttHostedServer : IMqttServer, IHostedService + public sealed class MqttHostedServer : MqttServer, IHostedService { - readonly MqttServer _mqttServer; - - public MqttHostedServer(MqttServerOptions options, IEnumerable adapters, IMqttNetLogger logger) + public MqttHostedServer(MqttServerOptions options, IEnumerable adapters, IMqttNetLogger logger) + : base(options, adapters, logger) { - _mqttServer = new MqttServer(options, adapters, logger); } public Task StartAsync(CancellationToken cancellationToken) { - _ = _mqttServer.StartAsync(); + _ = StartAsync(); return Task.CompletedTask; } public Task StopAsync(CancellationToken cancellationToken) { - return _mqttServer.StopAsync(); - } - - public Task PublishAsync(string senderClientId, MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) - { - return _mqttServer.PublishAsync(senderClientId, applicationMessage, cancellationToken); - } - - public void Dispose() - { - _mqttServer.Dispose(); - } - - public bool IsStarted => _mqttServer.IsStarted; - - public Task> GetClientStatusAsync() - { - return _mqttServer.GetClientStatusAsync(); - } - - public Task> GetSessionStatusAsync() - { - return _mqttServer.GetSessionStatusAsync(); - } - - public Task> GetRetainedApplicationMessagesAsync() - { - return _mqttServer.GetRetainedApplicationMessagesAsync(); - } - - public Task ClearRetainedApplicationMessagesAsync() - { - return _mqttServer.ClearRetainedApplicationMessagesAsync(); - } - - public Task SubscribeAsync(string clientId, ICollection topicFilters) - { - return _mqttServer.SubscribeAsync(clientId, topicFilters); - } - - public Task UnsubscribeAsync(string clientId, ICollection topicFilters) - { - return _mqttServer.UnsubscribeAsync(clientId, topicFilters); - } - - public Task StartAsync() - { - return _mqttServer.StartAsync(); - } - - public Task StopAsync() - { - return _mqttServer.StopAsync(); + return StopAsync(); } } } \ No newline at end of file diff --git a/Source/MQTTnet/MQTTnet.csproj.DotSettings b/Source/MQTTnet/MQTTnet.csproj.DotSettings index c72d42a0..67fc794f 100644 --- a/Source/MQTTnet/MQTTnet.csproj.DotSettings +++ b/Source/MQTTnet/MQTTnet.csproj.DotSettings @@ -1,3 +1,5 @@  + True True - True \ No newline at end of file + True + True \ No newline at end of file diff --git a/Source/MQTTnet/Server/InterceptingMqttClientPublishEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingMqttClientPublishEventArgs.cs similarity index 100% rename from Source/MQTTnet/Server/InterceptingMqttClientPublishEventArgs.cs rename to Source/MQTTnet/Server/Events/InterceptingMqttClientPublishEventArgs.cs diff --git a/Source/MQTTnet/Server/InterceptingMqttClientSubscriptionEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingMqttClientSubscriptionEventArgs.cs similarity index 100% rename from Source/MQTTnet/Server/InterceptingMqttClientSubscriptionEventArgs.cs rename to Source/MQTTnet/Server/Events/InterceptingMqttClientSubscriptionEventArgs.cs diff --git a/Source/MQTTnet/Server/InterceptingMqttClientUnsubscriptionEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingMqttClientUnsubscriptionEventArgs.cs similarity index 100% rename from Source/MQTTnet/Server/InterceptingMqttClientUnsubscriptionEventArgs.cs rename to Source/MQTTnet/Server/Events/InterceptingMqttClientUnsubscriptionEventArgs.cs diff --git a/Source/MQTTnet/Server/Events/InterceptingPacketEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingPacketEventArgs.cs new file mode 100644 index 00000000..4cecbd26 --- /dev/null +++ b/Source/MQTTnet/Server/Events/InterceptingPacketEventArgs.cs @@ -0,0 +1,30 @@ +using System; +using System.Threading; +using MQTTnet.Packets; + +namespace MQTTnet.Server +{ + public sealed class InterceptingPacketEventArgs : EventArgs + { + /// + /// Gets the client ID which has sent the packet or will receive the packet. + /// + public string ClientId { get; internal set; } + + /// + /// Gets the endpoint of the sending or receiving client. + /// + public string Endpoint { get; internal set; } + + /// + /// Gets or sets the MQTT packet which was received or will be sent. + /// + public MqttBasePacket Packet { get; set; } + + /// + /// Gets the cancellation token from the connection managing thread. + /// Use this in further event processing. + /// + public CancellationToken CancellationToken { get; internal set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttApplicationMessageNotConsumedEventArgs.cs b/Source/MQTTnet/Server/Events/MqttApplicationMessageNotConsumedEventArgs.cs similarity index 100% rename from Source/MQTTnet/Server/MqttApplicationMessageNotConsumedEventArgs.cs rename to Source/MQTTnet/Server/Events/MqttApplicationMessageNotConsumedEventArgs.cs diff --git a/Source/MQTTnet/Server/MqttClientConnectedEventArgs.cs b/Source/MQTTnet/Server/Events/MqttClientConnectedEventArgs.cs similarity index 100% rename from Source/MQTTnet/Server/MqttClientConnectedEventArgs.cs rename to Source/MQTTnet/Server/Events/MqttClientConnectedEventArgs.cs diff --git a/Source/MQTTnet/Server/MqttClientDisconnectedEventArgs.cs b/Source/MQTTnet/Server/Events/MqttClientDisconnectedEventArgs.cs similarity index 100% rename from Source/MQTTnet/Server/MqttClientDisconnectedEventArgs.cs rename to Source/MQTTnet/Server/Events/MqttClientDisconnectedEventArgs.cs diff --git a/Source/MQTTnet/Server/MqttClientSubscribedTopicEventArgs.cs b/Source/MQTTnet/Server/Events/MqttClientSubscribedTopicEventArgs.cs similarity index 100% rename from Source/MQTTnet/Server/MqttClientSubscribedTopicEventArgs.cs rename to Source/MQTTnet/Server/Events/MqttClientSubscribedTopicEventArgs.cs diff --git a/Source/MQTTnet/Server/MqttClientUnsubscribedTopicEventArgs.cs b/Source/MQTTnet/Server/Events/MqttClientUnsubscribedTopicEventArgs.cs similarity index 100% rename from Source/MQTTnet/Server/MqttClientUnsubscribedTopicEventArgs.cs rename to Source/MQTTnet/Server/Events/MqttClientUnsubscribedTopicEventArgs.cs diff --git a/Source/MQTTnet/Server/PreparingMqttClientSessionEventArgs.cs b/Source/MQTTnet/Server/Events/PreparingMqttClientSessionEventArgs.cs similarity index 100% rename from Source/MQTTnet/Server/PreparingMqttClientSessionEventArgs.cs rename to Source/MQTTnet/Server/Events/PreparingMqttClientSessionEventArgs.cs diff --git a/Source/MQTTnet/Server/ValidatingMqttClientConnectionEventArgs.cs b/Source/MQTTnet/Server/Events/ValidatingMqttClientConnectionEventArgs.cs similarity index 100% rename from Source/MQTTnet/Server/ValidatingMqttClientConnectionEventArgs.cs rename to Source/MQTTnet/Server/Events/ValidatingMqttClientConnectionEventArgs.cs diff --git a/Source/MQTTnet/Server/IMqttServer.cs b/Source/MQTTnet/Server/IMqttServer.cs deleted file mode 100644 index 389a9f20..00000000 --- a/Source/MQTTnet/Server/IMqttServer.cs +++ /dev/null @@ -1,23 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Threading.Tasks; - -namespace MQTTnet.Server -{ - public interface IMqttServer : IDisposable - { - bool IsStarted { get; } - - Task> GetClientStatusAsync(); - Task> GetSessionStatusAsync(); - - Task> GetRetainedApplicationMessagesAsync(); - Task ClearRetainedApplicationMessagesAsync(); - - Task SubscribeAsync(string clientId, ICollection topicFilters); - Task UnsubscribeAsync(string clientId, ICollection topicFilters); - - Task StartAsync(); - Task StopAsync(); - } -} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs index 90eda9d6..104bb870 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs @@ -27,6 +27,7 @@ namespace MQTTnet.Server.Internal readonly MqttApplicationMessageInterceptorInvoker _applicationMessageInterceptorInvoker; readonly MqttServerOptions _serverOptions; + readonly MqttServerEventContainer _eventContainer; readonly MqttConnectPacket _connectPacket; CancellationTokenSource _cancellationToken; @@ -41,6 +42,7 @@ namespace MQTTnet.Server.Internal IMqttNetLogger logger) { _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); + _eventContainer = eventContainer; _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager)); ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter)); @@ -155,6 +157,18 @@ namespace MQTTnet.Server.Internal while (!cancellationToken.IsCancellationRequested) { var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false); + + var interceptingPacketEventArgs = new InterceptingPacketEventArgs + { + ClientId = Id, + Endpoint = Endpoint, + Packet = packet, + CancellationToken = cancellationToken + }; + + await _eventContainer.InterceptingInboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false); + packet = interceptingPacketEventArgs.Packet; + if (packet == null) { // The client has closed the connection gracefully. diff --git a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs index 3d494079..3820377e 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs @@ -31,6 +31,10 @@ namespace MQTTnet.Server.Internal public AsyncEvent ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent(); + public AsyncEvent InterceptingInboundPacketEvent { get; } = new AsyncEvent(); + + public AsyncEvent InterceptingOutboundPacketEvent { get; } = new AsyncEvent(); + public AsyncEvent StartedEvent { get; } = new AsyncEvent(); public AsyncEvent StoppedEvent { get; } = new AsyncEvent(); diff --git a/Source/MQTTnet/Server/MqttApplicationMessageResponse.cs b/Source/MQTTnet/Server/MqttPublishResponse.cs similarity index 100% rename from Source/MQTTnet/Server/MqttApplicationMessageResponse.cs rename to Source/MQTTnet/Server/MqttPublishResponse.cs diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 4ff21c88..8f77263f 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -14,7 +14,7 @@ using MQTTnet.Server.Internal; namespace MQTTnet.Server { - public class MqttServer : Disposable, IMqttServer + public class MqttServer : Disposable { readonly MqttServerEventContainer _eventContainer; @@ -71,6 +71,18 @@ namespace MQTTnet.Server remove => _eventContainer.ClientDisconnectedEvent.RemoveHandler(value); } + public event Func InterceptInboundPacketAsync + { + add => _eventContainer.InterceptingInboundPacketEvent.AddHandler(value); + remove => _eventContainer.InterceptingInboundPacketEvent.RemoveHandler(value); + } + + public event Func InterceptingOutboundPacketAsync + { + add => _eventContainer.InterceptingOutboundPacketEvent.AddHandler(value); + remove => _eventContainer.InterceptingOutboundPacketEvent.RemoveHandler(value); + } + public event Func InterceptingClientSubscriptionAsync { add => _eventContainer.InterceptingClientSubscriptionEvent.AddHandler(value); diff --git a/Source/MQTTnet/Server/MqttServerExtensions.cs b/Source/MQTTnet/Server/MqttServerExtensions.cs index 60d9bc58..7a1f9ed8 100644 --- a/Source/MQTTnet/Server/MqttServerExtensions.cs +++ b/Source/MQTTnet/Server/MqttServerExtensions.cs @@ -8,7 +8,7 @@ namespace MQTTnet.Server { public static class MqttServerExtensions { - public static Task SubscribeAsync(this IMqttServer server, string clientId, params MqttTopicFilter[] topicFilters) + public static Task SubscribeAsync(this MqttServer server, string clientId, params MqttTopicFilter[] topicFilters) { if (server == null) throw new ArgumentNullException(nameof(server)); if (clientId == null) throw new ArgumentNullException(nameof(clientId)); @@ -17,7 +17,7 @@ namespace MQTTnet.Server return server.SubscribeAsync(clientId, topicFilters); } - public static Task SubscribeAsync(this IMqttServer server, string clientId, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel) + public static Task SubscribeAsync(this MqttServer server, string clientId, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel) { if (server == null) throw new ArgumentNullException(nameof(server)); if (clientId == null) throw new ArgumentNullException(nameof(clientId)); @@ -26,7 +26,7 @@ namespace MQTTnet.Server return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); } - public static Task SubscribeAsync(this IMqttServer server, string clientId, string topic) + public static Task SubscribeAsync(this MqttServer server, string clientId, string topic) { if (server == null) throw new ArgumentNullException(nameof(server)); if (clientId == null) throw new ArgumentNullException(nameof(clientId)); @@ -35,7 +35,7 @@ namespace MQTTnet.Server return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).Build()); } - public static Task UnsubscribeAsync(this IMqttServer server, string clientId, params string[] topicFilters) + public static Task UnsubscribeAsync(this MqttServer server, string clientId, params string[] topicFilters) { if (server == null) throw new ArgumentNullException(nameof(server)); if (clientId == null) throw new ArgumentNullException(nameof(clientId)); diff --git a/Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/Storage/IMqttRetainedMessagesManager.cs similarity index 87% rename from Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs rename to Source/MQTTnet/Server/Storage/IMqttRetainedMessagesManager.cs index bc4022e4..a3e752da 100644 --- a/Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/Storage/IMqttRetainedMessagesManager.cs @@ -1,5 +1,4 @@ -using MQTTnet.Diagnostics; -using System.Collections.Generic; +using System.Collections.Generic; using System.Threading.Tasks; using MQTTnet.Diagnostics.Logger; diff --git a/Source/MQTTnet/Server/IMqttServerPersistedSession.cs b/Source/MQTTnet/Server/Storage/IMqttServerPersistedSession.cs similarity index 100% rename from Source/MQTTnet/Server/IMqttServerPersistedSession.cs rename to Source/MQTTnet/Server/Storage/IMqttServerPersistedSession.cs diff --git a/Source/MQTTnet/Server/IMqttServerPersistedSessionsStorage.cs b/Source/MQTTnet/Server/Storage/IMqttServerPersistedSessionsStorage.cs similarity index 100% rename from Source/MQTTnet/Server/IMqttServerPersistedSessionsStorage.cs rename to Source/MQTTnet/Server/Storage/IMqttServerPersistedSessionsStorage.cs diff --git a/Source/MQTTnet/Server/IMqttServerStorage.cs b/Source/MQTTnet/Server/Storage/IMqttServerStorage.cs similarity index 100% rename from Source/MQTTnet/Server/IMqttServerStorage.cs rename to Source/MQTTnet/Server/Storage/IMqttServerStorage.cs diff --git a/Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs b/Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs index d000a239..afe11051 100644 --- a/Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs @@ -12,7 +12,7 @@ namespace MQTTnet.Benchmarks [MemoryDiagnoser] public sealed class MqttTcpChannelBenchmark { - IMqttServer _mqttServer; + MqttServer _mqttServer; IMqttChannel _serverChannel; IMqttChannel _clientChannel;