From 905306c3122b26ab92cbf5d90ff44d05202839c1 Mon Sep 17 00:00:00 2001 From: Christian Date: Tue, 26 Oct 2021 22:00:47 +0200 Subject: [PATCH] Add more handling options to interceptor contexts. --- Build/MQTTnet.nuspec | 2 + .../MqttConnectionContext.cs | 2 +- Source/MQTTnet.AspnetCore/MqttHostedServer.cs | 119 +++++++++++- .../ManagedMqttClient.cs | 11 +- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 8 +- .../Adapter/MqttPacketInspectorHandler.cs | 5 +- .../MqttClientDisconnectedEventArgs.cs | 2 + Source/MQTTnet/Client/IMqttClient.cs | 22 ++- Source/MQTTnet/Client/MqttClient.cs | 35 +++- Source/MQTTnet/Client/MqttClientExtensions.cs | 46 ++--- .../MqttClientSubscribeResultFactory.cs | 22 +-- .../MQTTnet/Formatter/IMqttPacketFormatter.cs | 5 +- Source/MQTTnet/Formatter/MqttPacketBuffer.cs | 51 +++++ .../Formatter/MqttPacketFormatterAdapter.cs | 2 +- .../Formatter/MqttPubAckPacketFactory.cs | 29 ++- .../Formatter/MqttPubRecPacketFactory.cs | 32 +++- .../Formatter/MqttPublishPacketFactory.cs | 15 +- .../Formatter/MqttSubAckPacketFactory.cs | 10 +- .../Formatter/MqttUnsubAckPacketFactory.cs | 6 +- .../Formatter/V3/MqttV310PacketFormatter.cs | 47 ++++- .../Formatter/V5/MqttV500PacketEncoder.cs | 27 ++- .../Formatter/V5/MqttV500PacketFormatter.cs | 2 +- .../MQTTnet/IApplicationMessagePublisher.cs | 2 +- Source/MQTTnet/IApplicationMessageReceiver.cs | 9 +- Source/MQTTnet/Internal/AsyncEvent.cs | 33 ++++ .../MQTTnet/MqttApplicationMessageBuilder.cs | 20 +- ...MqttApplicationMessageReceivedEventArgs.cs | 12 +- Source/MQTTnet/Packets/MqttSubAckPacket.cs | 11 +- ...qttApplicationMessageInterceptorInvoker.cs | 63 +++++++ .../Server/Internal/MqttClientConnection.cs | 116 +++++++----- .../Internal/MqttClientSessionsManager.cs | 175 +++++++++--------- .../MqttClientSubscriptionsManager.cs | 78 ++++---- .../Internal/MqttServerEventDispatcher.cs | 92 +++++---- ...scribeResult.cs => MqttSubscribeResult.cs} | 2 +- .../{Subscription.cs => MqttSubscription.cs} | 2 +- ...ribeResult.cs => MqttUnsubscribeResult.cs} | 2 +- ...qttApplicationMessageInterceptorContext.cs | 21 ++- .../Server/MqttApplicationMessageResponse.cs | 14 ++ ...qttApplicationMessageResponseReasonCode.cs | 17 ++ Source/MQTTnet/Server/MqttServer.cs | 66 ++++++- .../MQTTnet/Server/MqttSubscribeResponse.cs | 20 ++ .../MqttSubscriptionInterceptorContext.cs | 45 +---- .../MQTTnet/Server/MqttUnsubscribeResponse.cs | 19 ++ .../MqttUnsubscriptionInterceptorContext.cs | 19 +- .../ReaderExtensionsTest.cs | 2 +- .../ChannelAdapterBenchmark.cs | 2 +- .../MQTTnet.Benchmarks/SerializerBenchmark.cs | 2 +- .../Client/LowLevelMqttClient_Tests.cs | 2 +- .../Mockups/TestClientWrapper.cs | 18 ++ .../Mockups/TestServerWrapper.cs | 6 + .../MqttPacketSerializer_Tests.cs | 72 ++----- Tests/MQTTnet.Core.Tests/Server/General.cs | 2 +- .../Server/Session_Tests.cs | 2 +- .../Server/Subscribe_Tests.cs | 2 +- Tests/MQTTnet.TestApp.NetCore/ServerTest.cs | 4 +- 55 files changed, 943 insertions(+), 509 deletions(-) create mode 100644 Source/MQTTnet/Formatter/MqttPacketBuffer.cs create mode 100644 Source/MQTTnet/Internal/AsyncEvent.cs create mode 100644 Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs rename Source/MQTTnet/Server/Internal/{SubscribeResult.cs => MqttSubscribeResult.cs} (92%) rename Source/MQTTnet/Server/Internal/{Subscription.cs => MqttSubscription.cs} (92%) rename Source/MQTTnet/Server/Internal/{UnsubscribeResult.cs => MqttUnsubscribeResult.cs} (85%) create mode 100644 Source/MQTTnet/Server/MqttApplicationMessageResponse.cs create mode 100644 Source/MQTTnet/Server/MqttApplicationMessageResponseReasonCode.cs create mode 100644 Source/MQTTnet/Server/MqttSubscribeResponse.cs create mode 100644 Source/MQTTnet/Server/MqttUnsubscribeResponse.cs diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 47efd2ed..0c31c99b 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -14,6 +14,7 @@ * [Core] Added all builders to the MQTT factory. * [Core] Removed global logger and refactored logging (BREAKING CHANGE!). +* [Core] Improved memory management when working with large payloads. * [Client] Renamed _MqttClientConnectResult_ to _MqttClientConnectResult_ (BREAKING CHANGE!). * [Client] The OS will now choose the best TLS version to use. It is no longer fixed to 1.3 etc. (thanks to @patagonaa, #1271). * [Client] Exposed user properties and reason string in subscribe result. @@ -35,6 +36,7 @@ * [Server] Added support for returning individual subscription errors (#80 thanks to @jimch) * [Server] Improved topic filter comparisons (support for $). * [Server] Fixed wrong "IsSessionPresent" result (thanks to @logicaloud, #1268). +* [Server] Added more MQTTv5 response information to all interceptors (BREAKING CHANGE!). * [MQTTnet.Server] Moved server project to a dedicated GitHub repository. Git commit: $gitCommit diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index 95ea4711..305c9fa1 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -173,7 +173,7 @@ namespace MQTTnet.AspNetCore using (await _writerLock.WaitAsync(cancellationToken).ConfigureAwait(false)) { var buffer = formatter.Encode(packet); - var msg = buffer.AsMemory(); + var msg = buffer.ToArray().AsMemory(); var output = _output; var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false); if (result.IsCompleted) diff --git a/Source/MQTTnet.AspnetCore/MqttHostedServer.cs b/Source/MQTTnet.AspnetCore/MqttHostedServer.cs index 3a9c882d..9c08246a 100644 --- a/Source/MQTTnet.AspnetCore/MqttHostedServer.cs +++ b/Source/MQTTnet.AspnetCore/MqttHostedServer.cs @@ -4,30 +4,137 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Hosting; using MQTTnet.Adapter; +using MQTTnet.Client.Publishing; +using MQTTnet.Client.Receiving; using MQTTnet.Diagnostics.Logger; using MQTTnet.Server; +using MQTTnet.Server.Status; namespace MQTTnet.AspNetCore { - public sealed class MqttHostedServer : MqttServer, IHostedService + public sealed class MqttHostedServer : IMqttServer, IHostedService { + readonly MqttServer _mqttServer; readonly IMqttServerOptions _options; - public MqttHostedServer(IMqttServerOptions options, IEnumerable adapters, IMqttNetLogger logger) - : base(adapters, logger) + public MqttHostedServer(IMqttServerOptions options, IEnumerable adapters, IMqttNetLogger logger) { _options = options ?? throw new ArgumentNullException(nameof(options)); + + _mqttServer = new MqttServer(adapters, logger); } public Task StartAsync(CancellationToken cancellationToken) { - _ = StartAsync(_options); + _ = _mqttServer.StartAsync(_options); return Task.CompletedTask; } public Task StopAsync(CancellationToken cancellationToken) { - return StopAsync(); + return _mqttServer.StopAsync(); + } + + public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler + { + get => _mqttServer.ApplicationMessageReceivedHandler; + set => _mqttServer.ApplicationMessageReceivedHandler = value; + } + + public event Func ApplicationMessageReceivedAsync + { + add => _mqttServer.ApplicationMessageReceivedAsync += value; + remove => _mqttServer.ApplicationMessageReceivedAsync -= value; + } + + public Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) + { + return _mqttServer.PublishAsync(applicationMessage, cancellationToken); + } + + public void Dispose() + { + _mqttServer.Dispose(); + } + + public bool IsStarted => _mqttServer.IsStarted; + + public IMqttServerStartedHandler StartedHandler + { + get => _mqttServer.StartedHandler; + set => _mqttServer.StartedHandler = value; + } + + public IMqttServerStoppedHandler StoppedHandler + { + get => _mqttServer.StoppedHandler; + set => _mqttServer.StoppedHandler = value; + } + + public IMqttServerClientConnectedHandler ClientConnectedHandler + { + get => _mqttServer.ClientConnectedHandler; + set => _mqttServer.ClientConnectedHandler = value; + } + + public IMqttServerClientDisconnectedHandler ClientDisconnectedHandler + { + get => _mqttServer.ClientDisconnectedHandler; + set => _mqttServer.ClientDisconnectedHandler = value; + } + + public IMqttServerClientSubscribedTopicHandler ClientSubscribedTopicHandler + { + get => _mqttServer.ClientSubscribedTopicHandler; + set => _mqttServer.ClientSubscribedTopicHandler = value; + } + + public IMqttServerClientUnsubscribedTopicHandler ClientUnsubscribedTopicHandler + { + get => _mqttServer.ClientUnsubscribedTopicHandler; + set => _mqttServer.ClientUnsubscribedTopicHandler = value; + } + + public IMqttServerOptions Options => _mqttServer.Options; + + 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(IMqttServerOptions options) + { + return _mqttServer.StartAsync(options); + } + + public Task StopAsync() + { + return _mqttServer.StopAsync(); } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 6e1c1f54..608e5a72 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -17,10 +17,10 @@ using MQTTnet.Diagnostics.Logger; namespace MQTTnet.Extensions.ManagedClient { - public class ManagedMqttClient : Disposable, IManagedMqttClient + public sealed class ManagedMqttClient : Disposable, IManagedMqttClient { readonly BlockingQueue _messageQueue = new BlockingQueue(); - + /// /// The subscriptions are managed in 2 separate buckets: /// and are processed during normal operation @@ -29,7 +29,6 @@ namespace MQTTnet.Extensions.ManagedClient /// at reconnect and are solely owned by . /// readonly Dictionary _reconnectSubscriptions = new Dictionary(); - readonly Dictionary _subscriptions = new Dictionary(); readonly HashSet _unsubscriptions = new HashSet(); readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0); @@ -80,6 +79,12 @@ namespace MQTTnet.Extensions.ManagedClient set => InternalClient.ApplicationMessageReceivedHandler = value; } + public event Func ApplicationMessageReceivedAsync + { + add => InternalClient.ApplicationMessageReceivedAsync += value; + remove => InternalClient.ApplicationMessageReceivedAsync -= value; + } + public IApplicationMessageProcessedHandler ApplicationMessageProcessedHandler { get; set; } public IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; } diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 9b22973b..b7532a6e 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -116,11 +116,11 @@ namespace MQTTnet.Adapter var packetData = PacketFormatterAdapter.Encode(packet); _packetInspectorHandler.BeginSendPacket(packetData); - await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false); + await packetData.WriteToAsync(_channel, cancellationToken).ConfigureAwait(false); + + Interlocked.Add(ref _bytesReceived, packetData.Length); - Interlocked.Add(ref _bytesReceived, packetData.Count); - - _logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Count, packet); + _logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Length, packet); } catch (Exception exception) { diff --git a/Source/MQTTnet/Adapter/MqttPacketInspectorHandler.cs b/Source/MQTTnet/Adapter/MqttPacketInspectorHandler.cs index 04eb2603..828ea1ab 100644 --- a/Source/MQTTnet/Adapter/MqttPacketInspectorHandler.cs +++ b/Source/MQTTnet/Adapter/MqttPacketInspectorHandler.cs @@ -4,6 +4,7 @@ using System.Linq; using MQTTnet.Diagnostics; using MQTTnet.Diagnostics.Logger; using MQTTnet.Diagnostics.PacketInspection; +using MQTTnet.Formatter; namespace MQTTnet.Adapter { @@ -44,7 +45,7 @@ namespace MQTTnet.Adapter InspectPacket(buffer, MqttPacketFlowDirection.Inbound); } - public void BeginSendPacket(ArraySegment buffer) + public void BeginSendPacket(MqttPacketBuffer buffer) { if (_packetInspector == null) { @@ -56,7 +57,7 @@ namespace MQTTnet.Adapter // intended for debugging etc. so that this is OK. var bufferCopy = buffer.ToArray(); - InspectPacket(bufferCopy, MqttPacketFlowDirection.Outbound); + InspectPacket(bufferCopy.ToArray(), MqttPacketFlowDirection.Outbound); } public void FillReceiveBuffer(byte[] buffer) diff --git a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs index f4a47315..0fc7e490 100644 --- a/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs +++ b/Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs @@ -12,7 +12,9 @@ namespace MQTTnet.Client.Disconnecting ConnectResult = connectResult; Reason = reason; +#pragma warning disable 618 ReasonCode = reason; +#pragma warning restore 618 } public bool ClientWasConnected { get; } diff --git a/Source/MQTTnet/Client/IMqttClient.cs b/Source/MQTTnet/Client/IMqttClient.cs index 03bee88c..79feb1a3 100644 --- a/Source/MQTTnet/Client/IMqttClient.cs +++ b/Source/MQTTnet/Client/IMqttClient.cs @@ -22,17 +22,27 @@ namespace MQTTnet.Client /// IMqttClientConnectedHandler ConnectedHandler { get; set; } + /// + /// Fired when the client is connected. + /// + event Func ConnectedAsync; + /// /// Gets or sets the disconnected handler that is fired after the client has disconnected from the server. /// Hint: Initialize handlers before you connect the client to avoid issues. /// IMqttClientDisconnectedHandler DisconnectedHandler { get; set; } - Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken); + /// + /// Fired when the client is disconnected. + /// + event Func DisconnectedAsync; + + Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken = default); - Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken); + Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken = default); - Task PingAsync(CancellationToken cancellationToken); + Task PingAsync(CancellationToken cancellationToken = default); /// /// Sends extended authentication data. @@ -41,10 +51,10 @@ namespace MQTTnet.Client /// The extended data. /// A cancellation token to stop the task. /// A representing any asynchronous operation. - Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken); + Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default); - Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken); + Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default); - Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken); + Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 2e926fe9..2f66d100 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -21,13 +21,17 @@ using MQTTnet.Implementations; namespace MQTTnet.Client { - public class MqttClient : Disposable, IMqttClient + public sealed class MqttClient : Disposable, IMqttClient { readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); readonly object _disconnectLock = new object(); + readonly AsyncEvent _connectedEvent = new AsyncEvent(); + readonly AsyncEvent _disconnectedEvent = new AsyncEvent(); + readonly AsyncEvent _applicationMessageReceivedEvent = new AsyncEvent(); + readonly IMqttClientAdapterFactory _adapterFactory; readonly MqttNetSourceLogger _logger; @@ -56,10 +60,28 @@ namespace MQTTnet.Client public IMqttClientConnectedHandler ConnectedHandler { get; set; } + public event Func ConnectedAsync + { + add => _connectedEvent.AddHandler(value); + remove => _connectedEvent.RemoveHandler(value); + } + public IMqttClientDisconnectedHandler DisconnectedHandler { get; set; } + public event Func DisconnectedAsync + { + add => _disconnectedEvent.AddHandler(value); + remove => _disconnectedEvent.RemoveHandler(value); + } + public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; } + public event Func ApplicationMessageReceivedAsync + { + add => _applicationMessageReceivedEvent.AddHandler(value); + remove => _applicationMessageReceivedEvent.RemoveHandler(value); + } + public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected; public IMqttClientOptions Options { get; private set; } @@ -118,12 +140,15 @@ namespace MQTTnet.Client _logger.Info("Connected."); + var eventArgs = new MqttClientConnectedEventArgs(connectResult); var connectedHandler = ConnectedHandler; if (connectedHandler != null) { await connectedHandler.HandleConnectedAsync(new MqttClientConnectedEventArgs(connectResult)).ConfigureAwait(false); } + await _connectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + return connectResult; } catch (Exception exception) @@ -391,13 +416,17 @@ namespace MQTTnet.Client _logger.Info("Disconnected."); + var eventArgs = new MqttClientDisconnectedEventArgs(clientWasConnected, exception, connectResult, _disconnectReason); + var disconnectedHandler = DisconnectedHandler; if (disconnectedHandler != null) { // This handler must be executed in a new thread because otherwise a dead lock may happen // when trying to reconnect in that handler etc. - Task.Run(() => disconnectedHandler.HandleDisconnectedAsync(new MqttClientDisconnectedEventArgs(clientWasConnected, exception, connectResult, _disconnectReason))).RunInBackground(_logger); + Task.Run(() => disconnectedHandler.HandleDisconnectedAsync(eventArgs)).RunInBackground(_logger); } + + Task.Run(() => _disconnectedEvent.InvokeAsync(eventArgs)).RunInBackground(_logger); } } @@ -791,6 +820,8 @@ namespace MQTTnet.Client await handler.HandleApplicationMessageReceivedAsync(eventArgs).ConfigureAwait(false); } + await _applicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + return eventArgs; } diff --git a/Source/MQTTnet/Client/MqttClientExtensions.cs b/Source/MQTTnet/Client/MqttClientExtensions.cs index 117f3984..97c500b6 100644 --- a/Source/MQTTnet/Client/MqttClientExtensions.cs +++ b/Source/MQTTnet/Client/MqttClientExtensions.cs @@ -149,8 +149,19 @@ namespace MQTTnet.Client return client.DisconnectAsync(options, CancellationToken.None); } + + public static Task DisconnectAsync(this IMqttClient client, MqttClientDisconnectReason reasonCode, string reasonString = null) + { + if (client == null) throw new ArgumentNullException(nameof(client)); - public static Task DisconnectAsync(this IMqttClient client, CancellationToken cancellationToken) + return client.DisconnectAsync(new MqttClientDisconnectOptions + { + ReasonCode = reasonCode, + ReasonString = reasonString + }, CancellationToken.None); + } + + public static Task DisconnectAsync(this IMqttClient client, CancellationToken cancellationToken = default) { if (client == null) throw new ArgumentNullException(nameof(client)); @@ -194,43 +205,14 @@ namespace MQTTnet.Client return client.UnsubscribeAsync(options); } - - public static Task ConnectAsync(this IMqttClient client, IMqttClientOptions options) - { - if (client == null) throw new ArgumentNullException(nameof(client)); - - return client.ConnectAsync(options, CancellationToken.None); - } - + public static Task SendExtendedAuthenticationExchangeDataAsync(this IMqttClient client, MqttExtendedAuthenticationExchangeData data) { if (client == null) throw new ArgumentNullException(nameof(client)); return client.SendExtendedAuthenticationExchangeDataAsync(data, CancellationToken.None); } - - public static Task SubscribeAsync(this IMqttClient client, MqttClientSubscribeOptions options) - { - if (client == null) throw new ArgumentNullException(nameof(client)); - - return client.SubscribeAsync(options, CancellationToken.None); - } - - public static Task UnsubscribeAsync(this IMqttClient client, MqttClientUnsubscribeOptions options) - { - if (client == null) throw new ArgumentNullException(nameof(client)); - - return client.UnsubscribeAsync(options, CancellationToken.None); - } - - public static Task PublishAsync(this IMqttClient client, MqttApplicationMessage applicationMessage) - { - if (client == null) throw new ArgumentNullException(nameof(client)); - if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - - return client.PublishAsync(applicationMessage, CancellationToken.None); - } - + public static async Task PublishAsync(this IMqttClient client, IEnumerable applicationMessages) { if (client == null) throw new ArgumentNullException(nameof(client)); diff --git a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs index cec62e65..374e4d0a 100644 --- a/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs +++ b/Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs @@ -12,13 +12,6 @@ namespace MQTTnet.Client.Subscribing if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); if (subAckPacket == null) throw new ArgumentNullException(nameof(subAckPacket)); - // MQTTv3.1.1 handling. - if (subAckPacket.ReturnCodes.Any() && subAckPacket.ReturnCodes.Count != subscribePacket.TopicFilters.Count) - { - throw new MqttProtocolViolationException( - "The return codes are not matching the topic filters [MQTT-3.9.3-1]."); - } - // MQTTv5.0.0 handling. if (subAckPacket.ReasonCodes.Any() && subAckPacket.ReasonCodes.Count != subscribePacket.TopicFilters.Count) { @@ -43,19 +36,8 @@ namespace MQTTnet.Client.Subscribing static MqttClientSubscribeResultItem CreateSubscribeResultItem(int index, MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket) { - MqttClientSubscribeResultCode resultCode; - - if (subAckPacket.ReturnCodes.Any()) - { - // MQTTv3.1.1 handling. - resultCode = (MqttClientSubscribeResultCode) subAckPacket.ReturnCodes[index]; - } - else - { - // MQTTv5.0.0 handling. - resultCode = (MqttClientSubscribeResultCode) subAckPacket.ReasonCodes[index]; - } - + var resultCode = (MqttClientSubscribeResultCode) subAckPacket.ReasonCodes[index]; + return new MqttClientSubscribeResultItem { TopicFilter = subscribePacket.TopicFilters[index], diff --git a/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs b/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs index b4a26b84..b6137cea 100644 --- a/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs +++ b/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs @@ -1,12 +1,11 @@ -using System; -using MQTTnet.Adapter; +using MQTTnet.Adapter; using MQTTnet.Packets; namespace MQTTnet.Formatter { public interface IMqttPacketFormatter { - ArraySegment Encode(MqttBasePacket mqttPacket); + MqttPacketBuffer Encode(MqttBasePacket mqttPacket); MqttBasePacket Decode(ReceivedMqttPacket receivedMqttPacket); diff --git a/Source/MQTTnet/Formatter/MqttPacketBuffer.cs b/Source/MQTTnet/Formatter/MqttPacketBuffer.cs new file mode 100644 index 00000000..ec5e3ae7 --- /dev/null +++ b/Source/MQTTnet/Formatter/MqttPacketBuffer.cs @@ -0,0 +1,51 @@ +using System; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using MQTTnet.Channel; + +namespace MQTTnet.Formatter +{ + public sealed class MqttPacketBuffer + { + readonly ArraySegment[] _segments; + + public MqttPacketBuffer(params ArraySegment[] segments) + { + _segments = segments ?? throw new ArgumentNullException(nameof(segments)); + + Length = _segments.Sum(s => s.Count); + } + + public int Length { get; } + + public ArraySegment ToArray() + { + if (_segments.Length == 1) + { + return _segments[0]; + } + + using (var buffer = new MemoryStream(Length)) + { + foreach (var segment in _segments) + { + buffer.Write(segment.Array, segment.Offset, segment.Count); + } + + return new ArraySegment(buffer.ToArray()); + } + } + + public async Task WriteToAsync(IMqttChannel channel, CancellationToken cancellationToken) + { + if (channel == null) throw new ArgumentNullException(nameof(channel)); + + foreach (var segment in _segments) + { + await channel.WriteAsync(segment.Array, segment.Offset, segment.Count, cancellationToken).ConfigureAwait(false); + } + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs index bd7d3820..6cb82648 100644 --- a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs @@ -32,7 +32,7 @@ namespace MQTTnet.Formatter public IMqttPacketWriter Writer { get; } - public ArraySegment Encode(MqttBasePacket packet) + public MqttPacketBuffer Encode(MqttBasePacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); diff --git a/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs index ffc0f425..eb7ddc55 100644 --- a/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs @@ -1,25 +1,48 @@ using System; using MQTTnet.Packets; using MQTTnet.Protocol; +using MQTTnet.Server; namespace MQTTnet.Formatter { public sealed class MqttPubAckPacketFactory { + public MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageResponse applicationMessageResponse) + { + if (applicationMessageResponse == null) throw new ArgumentNullException(nameof(applicationMessageResponse)); + + if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + + var pubAckPacket = new MqttPubAckPacket + { + PacketIdentifier = publishPacket.PacketIdentifier, + ReasonCode = (MqttPubAckReasonCode) (int) applicationMessageResponse.ReasonCode, + Properties = + { + ReasonString = applicationMessageResponse.ReasonString + } + }; + + pubAckPacket.Properties.UserProperties.AddRange(applicationMessageResponse.UserProperties); + + return pubAckPacket; + } + public MqttPubAckPacket Create(MqttApplicationMessageReceivedEventArgs applicationMessageReceivedEventArgs) { if (applicationMessageReceivedEventArgs == null) throw new ArgumentNullException(nameof(applicationMessageReceivedEventArgs)); var pubAckPacket = Create(applicationMessageReceivedEventArgs.PublishPacket, applicationMessageReceivedEventArgs.ReasonCode); pubAckPacket.Properties.UserProperties.AddRange(applicationMessageReceivedEventArgs.ResponseUserProperties); - + pubAckPacket.Properties.ReasonString = applicationMessageReceivedEventArgs.ResponseReasonString; + return pubAckPacket; } - public MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode) + static MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode) { if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - + var pubAckPacket = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier, diff --git a/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs index 9fb369d0..b6cfc3aa 100644 --- a/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs @@ -1,9 +1,7 @@ - - - -using System; +using System; using MQTTnet.Packets; using MQTTnet.Protocol; +using MQTTnet.Server; namespace MQTTnet.Formatter { @@ -15,20 +13,38 @@ namespace MQTTnet.Formatter var pubRecPacket = Create(applicationMessageReceivedEventArgs.PublishPacket, applicationMessageReceivedEventArgs.ReasonCode); pubRecPacket.Properties.UserProperties.AddRange(applicationMessageReceivedEventArgs.ResponseUserProperties); - + return pubRecPacket; } - public MqttPubRecPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode) + public MqttBasePacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageResponse applicationMessageResponse) { if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - + + var pubRecPacket = new MqttPubRecPacket + { + PacketIdentifier = publishPacket.PacketIdentifier, + ReasonCode = (MqttPubRecReasonCode) (int) applicationMessageResponse.ReasonCode, + Properties = + { + ReasonString = applicationMessageResponse.ReasonString + } + }; + + pubRecPacket.Properties.UserProperties.AddRange(applicationMessageResponse.UserProperties); + return pubRecPacket; + } + + MqttPubRecPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode) + { + if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + var pubRecPacket = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier, ReasonCode = (MqttPubRecReasonCode) (int) applicationMessageReceivedReasonCode }; - + return pubRecPacket; } } diff --git a/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs index d0d98ec1..9d03e38e 100644 --- a/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs @@ -29,7 +29,7 @@ namespace MQTTnet.Formatter TopicAlias = applicationMessage.TopicAlias } }; - + if (applicationMessage.SubscriptionIdentifiers != null) { packet.Properties.SubscriptionIdentifiers.AddRange(applicationMessage.SubscriptionIdentifiers); @@ -73,5 +73,18 @@ namespace MQTTnet.Formatter return packet; } + + public MqttPublishPacket Clone(MqttPublishPacket publishPacket) + { + return new MqttPublishPacket + { + Topic = publishPacket.Topic, + Payload = publishPacket.Payload, + Retain = publishPacket.Retain, + QualityOfServiceLevel = publishPacket.QualityOfServiceLevel, + Dup = publishPacket.Dup, + PacketIdentifier = publishPacket.PacketIdentifier + }; + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs index 8396ba9b..856ef028 100644 --- a/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs @@ -6,21 +6,17 @@ namespace MQTTnet.Formatter { public sealed class MqttSubAckPacketFactory { - public MqttSubAckPacket Create(MqttSubscribePacket subscribePacket, SubscribeResult subscribeResult) + public MqttSubAckPacket Create(MqttSubscribePacket subscribePacket, MqttSubscribeResult mqttSubscribeResult) { if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); - if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult)); + if (mqttSubscribeResult == null) throw new ArgumentNullException(nameof(mqttSubscribeResult)); var subAckPacket = new MqttSubAckPacket { PacketIdentifier = subscribePacket.PacketIdentifier }; - // MQTTv3.1.1 - subAckPacket.ReturnCodes.AddRange(subscribeResult.ReturnCodes); - - // MQTTv5.0.0 - subAckPacket.ReasonCodes.AddRange(subscribeResult.ReasonCodes); + subAckPacket.ReasonCodes.AddRange(mqttSubscribeResult.ReasonCodes); return subAckPacket; } diff --git a/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs index e64ca0a3..9574facb 100644 --- a/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs @@ -6,10 +6,10 @@ namespace MQTTnet.Formatter { public sealed class MqttUnsubAckPacketFactory { - public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult) + public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, MqttUnsubscribeResult mqttUnsubscribeResult) { if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); - if (unsubscribeResult == null) throw new ArgumentNullException(nameof(unsubscribeResult)); + if (mqttUnsubscribeResult == null) throw new ArgumentNullException(nameof(mqttUnsubscribeResult)); var unsubAckPacket = new MqttUnsubAckPacket { @@ -17,7 +17,7 @@ namespace MQTTnet.Formatter }; // MQTTv5.0.0 only. - unsubAckPacket.ReasonCodes.AddRange(unsubscribeResult.ReasonCodes); + unsubAckPacket.ReasonCodes.AddRange(mqttUnsubscribeResult.ReasonCodes); return unsubAckPacket; } diff --git a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs index 0b526e05..98c738dc 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs @@ -21,7 +21,7 @@ namespace MQTTnet.Formatter.V3 _packetWriter = packetWriter; } - public ArraySegment Encode(MqttBasePacket packet) + public MqttPacketBuffer Encode(MqttBasePacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); @@ -32,6 +32,12 @@ namespace MQTTnet.Formatter.V3 var fixedHeader = EncodePacket(packet, _packetWriter); var remainingLength = (uint) (_packetWriter.Length - 5); + var publishPacket = packet as MqttPublishPacket; + if (publishPacket?.Payload != null) + { + remainingLength += (uint) publishPacket.Payload.Length; + } + var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength); var headerSize = FixedHeaderSize + remainingLengthSize; @@ -44,7 +50,15 @@ namespace MQTTnet.Formatter.V3 var buffer = _packetWriter.GetBuffer(); - return new ArraySegment(buffer, headerOffset, _packetWriter.Length - headerOffset); + var firstSegment = new ArraySegment(buffer, headerOffset, _packetWriter.Length - headerOffset); + + if (publishPacket?.Payload != null) + { + var payloadSegment = new ArraySegment(publishPacket.Payload, 0, publishPacket.Payload.Length); + return new MqttPacketBuffer(firstSegment, payloadSegment); + } + + return new MqttPacketBuffer(firstSegment); } public MqttBasePacket Decode(ReceivedMqttPacket receivedMqttPacket) @@ -300,7 +314,7 @@ namespace MQTTnet.Formatter.V3 while (!body.EndOfStream) { - packet.ReturnCodes.Add((MqttSubscribeReturnCode) body.ReadByte()); + packet.ReasonCodes.Add((MqttSubscribeReasonCode) body.ReadByte()); } return packet; @@ -442,10 +456,8 @@ namespace MQTTnet.Formatter.V3 } } - if (packet.Payload?.Length > 0) - { - packetWriter.Write(packet.Payload, 0, packet.Payload.Length); - } + // The payload is the past part of the packet. But it is not added here in order to keep + // memory allocation low. byte fixedHeader = 0; @@ -547,11 +559,26 @@ namespace MQTTnet.Formatter.V3 packetWriter.Write(packet.PacketIdentifier); - if (packet.ReturnCodes?.Any() == true) + if (packet.ReasonCodes.Any()) { - foreach (var packetSubscribeReturnCode in packet.ReturnCodes) + foreach (var packetSubscribeReturnCode in packet.ReasonCodes) { - packetWriter.Write((byte) packetSubscribeReturnCode); + if (packetSubscribeReturnCode == MqttSubscribeReasonCode.GrantedQoS0) + { + packetWriter.Write((byte) MqttSubscribeReturnCode.SuccessMaximumQoS0); + } + else if (packetSubscribeReturnCode == MqttSubscribeReasonCode.GrantedQoS1) + { + packetWriter.Write((byte) MqttSubscribeReturnCode.SuccessMaximumQoS1); + } + else if (packetSubscribeReturnCode == MqttSubscribeReasonCode.GrantedQoS2) + { + packetWriter.Write((byte) MqttSubscribeReturnCode.SuccessMaximumQoS2); + } + else + { + packetWriter.Write((byte)MqttSubscribeReturnCode.Failure); + } } } diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs index 2a640887..635d59e9 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs @@ -15,7 +15,7 @@ namespace MQTTnet.Formatter.V5 _packetWriter = packetWriter ?? throw new ArgumentNullException(nameof(packetWriter)); } - public ArraySegment Encode(MqttBasePacket packet) + public MqttPacketBuffer Encode(MqttBasePacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); @@ -24,8 +24,14 @@ namespace MQTTnet.Formatter.V5 _packetWriter.Seek(5); var fixedHeader = EncodePacket(packet, _packetWriter); - var remainingLength = (uint)(_packetWriter.Length - 5); + var remainingLength = (uint)_packetWriter.Length - 5; + var publishPacket = packet as MqttPublishPacket; + if (publishPacket?.Payload != null) + { + remainingLength += (uint)publishPacket.Payload.Length; + } + var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength); var headerSize = 1 + remainingLengthSize; @@ -37,7 +43,16 @@ namespace MQTTnet.Formatter.V5 _packetWriter.WriteVariableLengthInteger(remainingLength); var buffer = _packetWriter.GetBuffer(); - return new ArraySegment(buffer, headerOffset, _packetWriter.Length - headerOffset); + + var firstSegment = new ArraySegment(buffer, headerOffset, _packetWriter.Length - headerOffset); + + if (publishPacket?.Payload != null) + { + var payloadSegment = new ArraySegment(publishPacket.Payload, 0, publishPacket.Payload.Length); + return new MqttPacketBuffer(firstSegment, payloadSegment); + } + + return new MqttPacketBuffer(firstSegment); } public void FreeBuffer() @@ -256,10 +271,8 @@ namespace MQTTnet.Formatter.V5 propertiesWriter.WriteTo(packetWriter); - if (packet.Payload?.Length > 0) - { - packetWriter.Write(packet.Payload, 0, packet.Payload.Length); - } + // The payload is the past part of the packet. But it is not added here in order to keep + // memory allocation low. byte fixedHeader = 0; diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs index f5454963..a07c9a5a 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs @@ -14,7 +14,7 @@ namespace MQTTnet.Formatter.V5 _encoder = new MqttV500PacketEncoder(writer); } - public ArraySegment Encode(MqttBasePacket mqttPacket) + public MqttPacketBuffer Encode(MqttBasePacket mqttPacket) { if (mqttPacket == null) throw new ArgumentNullException(nameof(mqttPacket)); diff --git a/Source/MQTTnet/IApplicationMessagePublisher.cs b/Source/MQTTnet/IApplicationMessagePublisher.cs index bf17718c..7a93069c 100644 --- a/Source/MQTTnet/IApplicationMessagePublisher.cs +++ b/Source/MQTTnet/IApplicationMessagePublisher.cs @@ -6,6 +6,6 @@ namespace MQTTnet { public interface IApplicationMessagePublisher { - Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken); + Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default); } } diff --git a/Source/MQTTnet/IApplicationMessageReceiver.cs b/Source/MQTTnet/IApplicationMessageReceiver.cs index a44470d1..fbd20ccc 100644 --- a/Source/MQTTnet/IApplicationMessageReceiver.cs +++ b/Source/MQTTnet/IApplicationMessageReceiver.cs @@ -1,4 +1,6 @@ -using MQTTnet.Client.Receiving; +using System; +using System.Threading.Tasks; +using MQTTnet.Client.Receiving; namespace MQTTnet { @@ -9,5 +11,10 @@ namespace MQTTnet /// Hint: Initialize handlers before you connect the client to avoid issues. /// IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; } + + /// + /// Fired when an application message was received. + /// + event Func ApplicationMessageReceivedAsync; } } diff --git a/Source/MQTTnet/Internal/AsyncEvent.cs b/Source/MQTTnet/Internal/AsyncEvent.cs new file mode 100644 index 00000000..ba868f0e --- /dev/null +++ b/Source/MQTTnet/Internal/AsyncEvent.cs @@ -0,0 +1,33 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace MQTTnet.Internal +{ + public sealed class AsyncEvent + { + readonly List> _handlers = new List>(); + + public void AddHandler(Func handler) + { + if (handler == null) throw new ArgumentNullException(nameof(handler)); + + _handlers.Add(handler); + } + + public void RemoveHandler(Func handler) + { + if (handler == null) throw new ArgumentNullException(nameof(handler)); + + _handlers.Remove(handler); + } + + public async Task InvokeAsync(TEventArgs eventArgs) + { + foreach (var handler in _handlers) + { + await handler.Invoke(eventArgs); + } + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/MqttApplicationMessageBuilder.cs b/Source/MQTTnet/MqttApplicationMessageBuilder.cs index d494195f..879b7aa8 100644 --- a/Source/MQTTnet/MqttApplicationMessageBuilder.cs +++ b/Source/MQTTnet/MqttApplicationMessageBuilder.cs @@ -40,9 +40,7 @@ namespace MQTTnet /// The broker stores the last retained message and the corresponding QoS for that topic. /// bool _retain; - - bool _dup; - + /// /// The content type. /// The content type must be a UTF-8 encoded string. The content type value identifies the kind of UTF-8 encoded payload. @@ -198,13 +196,7 @@ namespace MQTTnet _retain = value; return this; } - - public MqttApplicationMessageBuilder WithDupFlag(bool value = true) - { - _dup = value; - return this; - } - + public MqttApplicationMessageBuilder WithAtLeastOnceQoS() { _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce; @@ -340,19 +332,13 @@ namespace MQTTnet { throw new MqttProtocolViolationException("A Topic Alias of 0 is not permitted. A sender MUST NOT send a PUBLISH packet containing a Topic Alias which has the value 0 [MQTT-3.3.2-8]."); } - - if (_qualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce && _dup) - { - throw new MqttProtocolViolationException("The DUP flag MUST be set to 0 for all QoS 0 messages [MQTT-3.3.1-2]."); - } - + var applicationMessage = new MqttApplicationMessage { Topic = _topic, Payload = _payload, QualityOfServiceLevel = _qualityOfServiceLevel, Retain = _retain, - Dup = _dup, ContentType = _contentType, ResponseTopic = _responseTopic, CorrelationData = _correlationData, diff --git a/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs b/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs index c3578a61..719630e3 100644 --- a/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs +++ b/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs @@ -24,7 +24,7 @@ namespace MQTTnet _acknowledgeHandler = acknowledgeHandler; } - internal MqttPublishPacket PublishPacket { get; } + internal MqttPublishPacket PublishPacket { get; set; } /// /// Gets the client identifier. @@ -36,13 +36,21 @@ namespace MQTTnet public bool ProcessingFailed { get; set; } + /// + /// Gets or sets the reason code which will be sent to the server. + /// public MqttApplicationMessageReceivedReasonCode ReasonCode { get; set; } = MqttApplicationMessageReceivedReasonCode.Success; /// - /// Gets or sets the user properties which being sent to the server in the ACK packet etc. + /// Gets or sets the user properties which will be sent to the server in the ACK packet etc. /// public List ResponseUserProperties { get; } = new List(); + /// + /// Gets or sets the reason string which will be sent to the server in the ACK packet. + /// + public string ResponseReasonString { get; set; } + /// /// Gets or sets whether this message was handled. /// This value can be used in user code for custom control flow. diff --git a/Source/MQTTnet/Packets/MqttSubAckPacket.cs b/Source/MQTTnet/Packets/MqttSubAckPacket.cs index c7c2242d..d307e5a3 100644 --- a/Source/MQTTnet/Packets/MqttSubAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttSubAckPacket.cs @@ -7,14 +7,10 @@ namespace MQTTnet.Packets public sealed class MqttSubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier { public ushort PacketIdentifier { get; set; } - - /// - /// Only available in MQTT v3.1.1. - /// - public List ReturnCodes { get; } = new List(); /// - /// Added in MQTTv5. + /// Reason Code is used in MQTTv5.0.0 and backward compatible to v.3.1.1. + /// Return Code is used in MQTTv3.1.1 /// public List ReasonCodes { get; } = new List(); @@ -25,10 +21,9 @@ namespace MQTTnet.Packets public override string ToString() { - var returnCodesText = string.Join(",", ReturnCodes.Select(f => f.ToString())); var reasonCodesText = string.Join(",", ReasonCodes.Select(f => f.ToString())); - return string.Concat("SubAck: [PacketIdentifier=", PacketIdentifier, "] [ReturnCodes=", returnCodesText, "] [ReasonCode=", reasonCodesText, "]"); + return string.Concat("SubAck: [PacketIdentifier=", PacketIdentifier, "] [ReasonCode=", reasonCodesText, "]"); } } } diff --git a/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs b/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs new file mode 100644 index 00000000..418d281f --- /dev/null +++ b/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs @@ -0,0 +1,63 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace MQTTnet.Server.Internal +{ + public sealed class MqttApplicationMessageInterceptorInvoker + { + readonly IMqttServerOptions _serverOptions; + readonly string _clientId; + readonly IDictionary _sessionItems; + + public MqttApplicationMessageInterceptorInvoker(IMqttServerOptions serverOptions, string clientId, IDictionary sessionItems) + { + _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); + _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); + _sessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); + } + + public MqttApplicationMessage ApplicationMessage { get; private set; } + + public MqttApplicationMessageResponse Response { get; private set; } = new MqttApplicationMessageResponse(); + + public bool ProcessPublish { get; private set; } = true; + + public bool CloseConnection { get; private set; } + + public async Task Invoke(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) + { + // Reset to defaults. + Response = new MqttApplicationMessageResponse(); + ProcessPublish = true; + CloseConnection = false; + ApplicationMessage = applicationMessage; + + // Intercept. + var interceptor = _serverOptions.ApplicationMessageInterceptor; + if (interceptor == null) + { + return; + } + + var interceptorContext = new MqttApplicationMessageInterceptorContext + { + ClientId = _clientId, + ApplicationMessage = applicationMessage, + SessionItems = _sessionItems, + ProcessPublish = true, + CloseConnection = false, + CancellationToken = cancellationToken + }; + + await interceptor.InterceptApplicationMessagePublishAsync(interceptorContext).ConfigureAwait(false); + + // Expose results. + ProcessPublish = interceptorContext.ProcessPublish && interceptorContext.ApplicationMessage != null; + CloseConnection = interceptorContext.CloseConnection; + Response = interceptorContext.Response; + ApplicationMessage = interceptorContext.ApplicationMessage; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs index 3355b99c..ff5dcc40 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs @@ -7,7 +7,6 @@ using MQTTnet.Client.Disconnecting; using MQTTnet.Diagnostics.Logger; using MQTTnet.Exceptions; using MQTTnet.Formatter; -using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.PacketDispatcher; using MQTTnet.Packets; @@ -20,16 +19,18 @@ namespace MQTTnet.Server.Internal readonly Dictionary _topicAlias = new Dictionary(); readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); + readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory(); readonly MqttClientSessionsManager _sessionsManager; - readonly MqttNetSourceLogger _logger; + + readonly MqttApplicationMessageInterceptorInvoker _applicationMessageInterceptorInvoker; + readonly IMqttServerOptions _serverOptions; - readonly MqttConnectPacket _connectPacket; CancellationTokenSource _cancellationToken; - + public MqttClientConnection( MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, @@ -47,6 +48,8 @@ namespace MQTTnet.Server.Internal Session = session ?? throw new ArgumentNullException(nameof(session)); _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket)); + _applicationMessageInterceptorInvoker = new MqttApplicationMessageInterceptorInvoker(_serverOptions, Id, Session.Items); + if (logger == null) throw new ArgumentNullException(nameof(logger)); _logger = logger.WithSource(nameof(MqttClientConnection)); } @@ -109,10 +112,7 @@ namespace MQTTnet.Server.Internal try { Task.Run(() => SendPacketsLoop(cancellationToken.Token), cancellationToken.Token).RunInBackground(_logger); - - // TODO: Check!. - //Session.IsCleanSession = false; - + IsRunning = true; await ReceivePackagesLoop(cancellationToken.Token).ConfigureAwait(false); @@ -130,8 +130,9 @@ namespace MQTTnet.Server.Internal if (!IsTakenOver && !IsCleanDisconnect && Session.LatestConnectPacket.WillFlag && !Session.WillMessageSent) { - var publishPacket = _packetFactories.Publish.Create(Session.LatestConnectPacket); - _sessionsManager.DispatchPublishPacket(publishPacket, this); + var willPublishPacket = _packetFactories.Publish.Create(Session.LatestConnectPacket); + var willApplicationMessage = _applicationMessageFactory.Create(willPublishPacket); + _ = _sessionsManager.DispatchPublishPacket(willApplicationMessage, this); Session.WillMessageSent = true; } @@ -163,7 +164,7 @@ namespace MQTTnet.Server.Internal if (packet is MqttPublishPacket publishPacket) { - HandleIncomingPublishPacket(publishPacket); + await HandleIncomingPublishPacket(publishPacket, cancellationToken).ConfigureAwait(false); } else if (packet is MqttPubAckPacket pubAckPacket) { @@ -329,7 +330,7 @@ namespace MQTTnet.Server.Internal Session.EnqueuePacket(new MqttPacketBusItem(publishPacket)); } } - + StopInternal(); } } @@ -366,7 +367,7 @@ namespace MQTTnet.Server.Internal { _cancellationToken?.Cancel(); } - + void HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket) { var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); @@ -413,12 +414,27 @@ namespace MQTTnet.Server.Internal StopInternal(); } } - - void HandleIncomingPublishPacket(MqttPublishPacket publishPacket) + + async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken) { HandleTopicAlias(publishPacket); - _sessionsManager.DispatchPublishPacket(publishPacket, this); + var applicationMessage = new MqttApplicationMessageFactory().Create(publishPacket); + + await _applicationMessageInterceptorInvoker.Invoke(applicationMessage, cancellationToken).ConfigureAwait(false); + + applicationMessage = _applicationMessageInterceptorInvoker.ApplicationMessage; + + if (_applicationMessageInterceptorInvoker.CloseConnection) + { + await StopAsync(MqttClientDisconnectReason.UnspecifiedError); + return; + } + + if (_applicationMessageInterceptorInvoker.ProcessPublish) + { + await _sessionsManager.DispatchPublishPacket(applicationMessage, this).ConfigureAwait(false); + } switch (publishPacket.QualityOfServiceLevel) { @@ -429,13 +445,13 @@ namespace MQTTnet.Server.Internal } case MqttQualityOfServiceLevel.AtLeastOnce: { - var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, _applicationMessageInterceptorInvoker.Response); Session.EnqueuePacket(new MqttPacketBusItem(pubAckPacket)); break; } case MqttQualityOfServiceLevel.ExactlyOnce: { - var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success); + var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, _applicationMessageInterceptorInvoker.Response); Session.EnqueuePacket(new MqttPacketBusItem(pubRecPacket)); break; } @@ -497,37 +513,39 @@ namespace MQTTnet.Server.Internal } } - // TODO: Fix! - async Task InvokeClientMessageQueueInterceptor(MqttPublishPacket publishPacket, MqttQueuedApplicationMessage queuedApplicationMessage) - { - if (_serverOptions.ClientMessageQueueInterceptor == null) - { - return publishPacket; - } + // // TODO: Fix! + // async Task InvokeClientMessageQueueInterceptor(MqttPublishPacket publishPacket, MqttQueuedApplicationMessage queuedApplicationMessage) + // { + // if (_serverOptions.ClientMessageQueueInterceptor == null) + // { + // return publishPacket; + // } + // + // var context = new MqttClientMessageQueueInterceptorContext + // { + // SenderClientId = queuedApplicationMessage.SenderClientId, + // ReceiverClientId = Id, + // ApplicationMessage = queuedApplicationMessage.ApplicationMessage, + // SubscriptionQualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel + // }; + // + // if (_serverOptions.ClientMessageQueueInterceptor != null) + // { + // await _serverOptions.ClientMessageQueueInterceptor.InterceptClientMessageQueueEnqueueAsync(context).ConfigureAwait(false); + // } + // + // if (!context.AcceptEnqueue || context.ApplicationMessage == null) + // { + // return null; + // } + // + // publishPacket.Topic = context.ApplicationMessage.Topic; + // publishPacket.Payload = context.ApplicationMessage.Payload; + // publishPacket.QualityOfServiceLevel = context.SubscriptionQualityOfServiceLevel; + // + // return publishPacket; + // } + // - var context = new MqttClientMessageQueueInterceptorContext - { - SenderClientId = queuedApplicationMessage.SenderClientId, - ReceiverClientId = Id, - ApplicationMessage = queuedApplicationMessage.ApplicationMessage, - SubscriptionQualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel - }; - - if (_serverOptions.ClientMessageQueueInterceptor != null) - { - await _serverOptions.ClientMessageQueueInterceptor.InterceptClientMessageQueueEnqueueAsync(context).ConfigureAwait(false); - } - - if (!context.AcceptEnqueue || context.ApplicationMessage == null) - { - return null; - } - - publishPacket.Topic = context.ApplicationMessage.Topic; - publishPacket.Payload = context.ApplicationMessage.Payload; - publishPacket.QualityOfServiceLevel = context.SubscriptionQualityOfServiceLevel; - - return publishPacket; - } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 8b10367c..9a1de8b4 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -16,9 +16,9 @@ using MQTTnet.Server.Status; namespace MQTTnet.Server.Internal { - public sealed class MqttClientSessionsManager : IDisposable + public sealed class MqttClientSessionsManager { - readonly BlockingCollection _messageQueue = new BlockingCollection(); + //readonly BlockingCollection _messageQueue = new BlockingCollection(); readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); readonly AsyncLock _createConnectionSyncRoot = new AsyncLock(); @@ -51,10 +51,10 @@ namespace MQTTnet.Server.Internal _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager)); } - public void Start(CancellationToken cancellationToken) - { - Task.Run(() => TryProcessQueuedApplicationMessagesAsync(cancellationToken), cancellationToken).RunInBackground(_logger); - } + // public void Start(CancellationToken cancellationToken) + // { + // Task.Run(() => TryProcessQueuedApplicationMessagesAsync(cancellationToken), cancellationToken).RunInBackground(_logger); + // } async Task ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken) { @@ -225,13 +225,13 @@ namespace MQTTnet.Server.Internal return Task.FromResult((IList)result); } - public void DispatchPublishPacket(MqttPublishPacket publishPacket, MqttClientConnection sender) - { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - - _messageQueue.Add(new MqttPendingApplicationMessage(publishPacket, sender)); - } - + // public void DispatchPublishPacket(MqttPublishPacket publishPacket, MqttClientConnection sender) + // { + // if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + // + // _messageQueue.Add(new MqttPendingApplicationMessage(publishPacket, sender)); + // } + public async Task SubscribeAsync(string clientId, ICollection topicFilters) { if (clientId == null) throw new ArgumentNullException(nameof(clientId)); @@ -304,80 +304,82 @@ namespace MQTTnet.Server.Internal _logger.Verbose("Session for client '{0}' deleted.", clientId); } - public void Dispose() - { - _messageQueue?.Dispose(); - } + // public void Dispose() + // { + // _messageQueue?.Dispose(); + // } - async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken) - { - // Make sure all queued messages are processed before server stops. - while (!cancellationToken.IsCancellationRequested || _messageQueue.Any()) - { - try - { - await TryProcessNextQueuedApplicationMessage().ConfigureAwait(false); - } - catch (OperationCanceledException) - { - } - catch (Exception exception) - { - _logger.Error(exception, "Unhandled exception while processing queued application messages."); - } - } - } + // async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken) + // { + // // Make sure all queued messages are processed before server stops. + // while (!cancellationToken.IsCancellationRequested || _messageQueue.Any()) + // { + // try + // { + // await TryProcessNextQueuedApplicationMessage().ConfigureAwait(false); + // } + // catch (OperationCanceledException) + // { + // } + // catch (Exception exception) + // { + // _logger.Error(exception, "Unhandled exception while processing queued application messages."); + // } + // } + // } - async Task TryProcessNextQueuedApplicationMessage() + public async Task DispatchPublishPacket(MqttApplicationMessage applicationMessage, MqttClientConnection sender) { try { - MqttPendingApplicationMessage pendingApplicationMessage; - try - { - pendingApplicationMessage = _messageQueue.Take(); - } - catch (ArgumentNullException) - { - return; - } - // catch (ThreadAbortException) + var senderClientId = sender?.Id ?? _options.ClientId; + + // MqttPendingApplicationMessage pendingApplicationMessage; + // try + // { + // pendingApplicationMessage = _messageQueue.Take(); + // } + // catch (ArgumentNullException) + // { + // return; + // } + // // catch (ThreadAbortException) + // // { + // // return; + // // } + // catch (ObjectDisposedException) // { // return; // } - catch (ObjectDisposedException) - { - return; - } - var clientConnection = pendingApplicationMessage.Sender; - var senderClientId = clientConnection?.Id ?? _options.ClientId; + //var clientConnection = pendingApplicationMessage.Sender; + //var senderClientId = clientConnection?.Id ?? _options.ClientId; - // TODO: Only use PUBLISH packets internally! - var applicationMessage = new MqttApplicationMessageFactory().Create(pendingApplicationMessage.PublishPacket); - - var interceptor = _options.ApplicationMessageInterceptor; - if (interceptor != null) - { - var interceptorContext = await InterceptApplicationMessageAsync(interceptor, clientConnection, applicationMessage).ConfigureAwait(false); - if (interceptorContext != null) - { - if (interceptorContext.CloseConnection) - { - if (clientConnection != null) - { - await clientConnection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); - } - } - - if (interceptorContext.ApplicationMessage == null || !interceptorContext.ProcessPublish) - { - return; - } - - applicationMessage = interceptorContext.ApplicationMessage; - } - } + // // TODO: Only use PUBLISH packets internally! + // var applicationMessage = new MqttApplicationMessageFactory().Create(pendingApplicationMessage.PublishPacket); + // + // var interceptor = _options.ApplicationMessageInterceptor; + // if (interceptor != null) + // { + // var interceptorContext = await InterceptApplicationMessageAsync(interceptor, clientConnection, applicationMessage).ConfigureAwait(false); + // if (interceptorContext != null) + // { + // if (interceptorContext.CloseConnection) + // { + // if (clientConnection != null) + // { + // await clientConnection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); + // } + // } + // + // if (interceptorContext.ApplicationMessage == null || !interceptorContext.ProcessPublish) + // { + // return; + // } + // + // applicationMessage = interceptorContext.ApplicationMessage; + // } + // } await _eventDispatcher.SafeNotifyApplicationMessageReceivedAsync(senderClientId, applicationMessage).ConfigureAwait(false); @@ -414,28 +416,27 @@ namespace MQTTnet.Server.Internal // SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers // }; - var publishPacket = _packetFactories.Publish.Create(applicationMessage); - - publishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel; - publishPacket.Properties.SubscriptionIdentifiers.AddRange(checkSubscriptionsResult.SubscriptionIdentifiers); + var newPublishPacket = _packetFactories.Publish.Create(applicationMessage); + newPublishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel; + newPublishPacket.Properties.SubscriptionIdentifiers.AddRange(checkSubscriptionsResult.SubscriptionIdentifiers); - if (publishPacket.QualityOfServiceLevel > 0) + if (newPublishPacket.QualityOfServiceLevel > 0) { - publishPacket.PacketIdentifier = clientSession.PacketIdentifierProvider.GetNextPacketIdentifier(); + newPublishPacket.PacketIdentifier = clientSession.PacketIdentifierProvider.GetNextPacketIdentifier(); } if (checkSubscriptionsResult.RetainAsPublished) { // Transfer the original retain state from the publisher. // This is a MQTTv5 feature. - publishPacket.Retain = applicationMessage.Retain; + newPublishPacket.Retain = applicationMessage.Retain; } else { - publishPacket.Retain = false; + newPublishPacket.Retain = false; } - clientSession.EnqueuePacket(new MqttPacketBusItem(publishPacket)); + clientSession.EnqueuePacket(new MqttPacketBusItem(newPublishPacket)); deliveryCount++; } @@ -448,7 +449,7 @@ namespace MQTTnet.Server.Internal } // The delegate signature is the same as for regular message interceptor. So the call is fine and just uses a different interceptor. - await InterceptApplicationMessageAsync(undeliveredMessageInterceptor, clientConnection, applicationMessage).ConfigureAwait(false); + await InterceptApplicationMessageAsync(undeliveredMessageInterceptor, sender, applicationMessage).ConfigureAwait(false); } } catch (Exception exception) diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 6695b442..b65f7a9a 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -15,7 +15,7 @@ namespace MQTTnet.Server.Internal // Since writing is done by multiple threads (server API or connection thread), we cannot avoid locking // completely by swapping references etc. readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim(); - readonly Dictionary _subscriptions = new Dictionary(4096); + readonly Dictionary _subscriptions = new Dictionary(4096); readonly MqttClientSession _clientSession; readonly IMqttServerOptions _options; @@ -34,12 +34,12 @@ namespace MQTTnet.Server.Internal _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager)); } - public async Task Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) + public async Task Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) { if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); var retainedApplicationMessages = await _retainedMessagesManager.GetMessagesAsync().ConfigureAwait(false); - var result = new SubscribeResult(); + var result = new MqttSubscribeResult(); // The topic filters are order by its QoS so that the higher QoS will win over a // lower one. @@ -47,18 +47,18 @@ namespace MQTTnet.Server.Internal { var interceptorContext = await InterceptSubscribe(originalTopicFilter, cancellationToken).ConfigureAwait(false); var finalTopicFilter = interceptorContext.TopicFilter; - var acceptSubscription = interceptorContext.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2; + var processSubscription = interceptorContext.Response.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2; - if (string.IsNullOrEmpty(finalTopicFilter.Topic) || !acceptSubscription) + if (string.IsNullOrEmpty(finalTopicFilter.Topic) || !processSubscription) { // Return codes is for MQTT < 5 and Reason Code is for MQTT > 5. result.ReturnCodes.Add(MqttSubscribeReturnCode.Failure); - result.ReasonCodes.Add(interceptorContext.ReasonCode); + result.ReasonCodes.Add(interceptorContext.Response.ReasonCode); } else { - result.ReturnCodes.Add(interceptorContext.ReturnCode); - result.ReasonCodes.Add(interceptorContext.ReasonCode); + result.ReturnCodes.Add((MqttSubscribeReturnCode)interceptorContext.Response.ReasonCode); + result.ReasonCodes.Add(interceptorContext.Response.ReasonCode); } if (interceptorContext.CloseConnection) @@ -68,7 +68,7 @@ namespace MQTTnet.Server.Internal result.CloseConnection = true; } - if (!acceptSubscription || string.IsNullOrEmpty(finalTopicFilter.Topic)) + if (!processSubscription || string.IsNullOrEmpty(finalTopicFilter.Topic)) { continue; } @@ -78,7 +78,7 @@ namespace MQTTnet.Server.Internal var subscription = CreateSubscription( finalTopicFilter, subscribePacket.Properties?.SubscriptionIdentifier ?? 0, - interceptorContext.ReasonCode); + interceptorContext.Response.ReasonCode); await _eventDispatcher .SafeNotifyClientSubscribedTopicAsync(_clientSession.Id, finalTopicFilter) @@ -91,11 +91,11 @@ namespace MQTTnet.Server.Internal return result; } - public async Task Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) + public async Task Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket)); - var result = new UnsubscribeResult(); + var result = new MqttUnsubscribeResult(); _subscriptionsLock.EnterWriteLock(); try @@ -105,9 +105,9 @@ namespace MQTTnet.Server.Internal _subscriptions.TryGetValue(topicFilter, out var existingSubscription); var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription, cancellationToken).ConfigureAwait(false); - var acceptUnsubscription = interceptorContext.ReasonCode == MqttUnsubscribeReasonCode.Success; + var acceptUnsubscription = interceptorContext.Response.ReasonCode == MqttUnsubscribeReasonCode.Success; - result.ReasonCodes.Add(interceptorContext.ReasonCode); + result.ReasonCodes.Add(interceptorContext.Response.ReasonCode); if (interceptorContext.CloseConnection) { @@ -143,7 +143,7 @@ namespace MQTTnet.Server.Internal public CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel qosLevel, string senderClientId) { - List subscriptions; + List subscriptions; _subscriptionsLock.EnterReadLock(); try { @@ -201,10 +201,9 @@ namespace MQTTnet.Server.Internal }; } - Subscription CreateSubscription(MqttTopicFilter topicFilter, uint subscriptionIdentifier, - MqttSubscribeReasonCode reasonCode) + MqttSubscription CreateSubscription(MqttTopicFilter topicFilter, uint subscriptionIdentifier, MqttSubscribeReasonCode reasonCode) { - var subscription = new Subscription + var subscription = new MqttSubscription { Topic = topicFilter.Topic, NoLocal = topicFilter.NoLocal, @@ -245,7 +244,7 @@ namespace MQTTnet.Server.Internal } static void FilterRetainedApplicationMessages(IList retainedApplicationMessages, - Subscription subscription, SubscribeResult subscribeResult) + MqttSubscription mqttSubscription, MqttSubscribeResult mqttSubscribeResult) { for (var i = retainedApplicationMessages.Count - 1; i >= 0; i--) { @@ -255,20 +254,20 @@ namespace MQTTnet.Server.Internal continue; } - if (subscription.RetainHandling == MqttRetainHandling.DoNotSendOnSubscribe) + if (mqttSubscription.RetainHandling == MqttRetainHandling.DoNotSendOnSubscribe) { // This is a MQTT V5+ feature. continue; } - if (subscription.RetainHandling == MqttRetainHandling.SendAtSubscribeIfNewSubscriptionOnly && - !subscription.IsNewSubscription) + if (mqttSubscription.RetainHandling == MqttRetainHandling.SendAtSubscribeIfNewSubscriptionOnly && + !mqttSubscription.IsNewSubscription) { // This is a MQTT V5+ feature. continue; } - if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, subscription.Topic) != + if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, mqttSubscription.Topic) != MqttTopicFilterCompareResult.IsMatch) { continue; @@ -278,15 +277,15 @@ namespace MQTTnet.Server.Internal { ApplicationMessage = retainedApplicationMessage, IsRetainedMessage = true, - SubscriptionQualityOfServiceLevel = subscription.GrantedQualityOfServiceLevel + SubscriptionQualityOfServiceLevel = mqttSubscription.GrantedQualityOfServiceLevel }; - if (subscription.Identifier > 0) + if (mqttSubscription.Identifier > 0) { - queuedApplicationMessage.SubscriptionIdentifiers = new List {subscription.Identifier}; + queuedApplicationMessage.SubscriptionIdentifiers = new List {mqttSubscription.Identifier}; } - subscribeResult.RetainedApplicationMessages.Add(queuedApplicationMessage); + mqttSubscribeResult.RetainedApplicationMessages.Add(queuedApplicationMessage); retainedApplicationMessages[i] = null; } @@ -305,27 +304,20 @@ namespace MQTTnet.Server.Internal if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) { - context.ReturnCode = MqttSubscribeReturnCode.SuccessMaximumQoS0; - context.ReasonCode = MqttSubscribeReasonCode.GrantedQoS0; + context.Response.ReasonCode = MqttSubscribeReasonCode.GrantedQoS0; } else if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) { - context.ReturnCode = MqttSubscribeReturnCode.SuccessMaximumQoS1; - context.ReasonCode = MqttSubscribeReasonCode.GrantedQoS1; + context.Response.ReasonCode = MqttSubscribeReasonCode.GrantedQoS1; } else if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) { - context.ReturnCode = MqttSubscribeReturnCode.SuccessMaximumQoS2; - context.ReasonCode = MqttSubscribeReasonCode.GrantedQoS2; + context.Response.ReasonCode = MqttSubscribeReasonCode.GrantedQoS2; } - - context.DefaultReturnCode = context.ReturnCode; - context.DefaultReasonCode = context.ReasonCode; - + if (topicFilter.Topic.StartsWith("$share/")) { - context.ReasonCode = MqttSubscribeReasonCode.SharedSubscriptionsNotSupported; - context.ReturnCode = MqttSubscribeReturnCode.Failure; + context.Response.ReasonCode = MqttSubscribeReasonCode.SharedSubscriptionsNotSupported; } else { @@ -339,7 +331,7 @@ namespace MQTTnet.Server.Internal return context; } - async Task InterceptUnsubscribe(string topicFilter, Subscription subscription, CancellationToken cancellationToken) + async Task InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken) { var context = new MqttUnsubscriptionInterceptorContext { @@ -349,13 +341,13 @@ namespace MQTTnet.Server.Internal CancellationToken = cancellationToken }; - if (subscription == null) + if (mqttSubscription == null) { - context.DefaultReasonCode = MqttUnsubscribeReasonCode.NoSubscriptionExisted; + context.Response.ReasonCode = MqttUnsubscribeReasonCode.NoSubscriptionExisted; } else { - context.DefaultReasonCode = MqttUnsubscribeReasonCode.Success; + context.Response.ReasonCode = MqttUnsubscribeReasonCode.Success; } var interceptor = _options.UnsubscriptionInterceptor; diff --git a/Source/MQTTnet/Server/Internal/MqttServerEventDispatcher.cs b/Source/MQTTnet/Server/Internal/MqttServerEventDispatcher.cs index 97b90251..3410bd42 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerEventDispatcher.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerEventDispatcher.cs @@ -2,8 +2,8 @@ using System; using System.Threading.Tasks; using MQTTnet.Adapter; using MQTTnet.Client.Receiving; -using MQTTnet.Diagnostics; using MQTTnet.Diagnostics.Logger; +using MQTTnet.Internal; using MQTTnet.Packets; namespace MQTTnet.Server.Internal @@ -19,6 +19,16 @@ namespace MQTTnet.Server.Internal _logger = logger.WithSource(nameof(MqttServerEventDispatcher)); } + public AsyncEvent ClientConnectedEvent { get; } = new AsyncEvent(); + + public AsyncEvent ClientDisconnectedEvent { get; } = new AsyncEvent(); + + public AsyncEvent ClientSubscribedTopicEvent { get; } = new AsyncEvent(); + + public AsyncEvent ClientUnsubscribedTopicEvent { get; } = new AsyncEvent(); + + public AsyncEvent ApplicationMessageReceivedEvent { get; } = new AsyncEvent(); + public IMqttServerClientConnectedHandler ClientConnectedHandler { get; set; } public IMqttServerClientDisconnectedHandler ClientDisconnectedHandler { get; set; } @@ -33,19 +43,21 @@ namespace MQTTnet.Server.Internal { try { - var handler = ClientConnectedHandler; - if (handler == null) - { - return; - } - - await handler.HandleClientConnectedAsync(new MqttServerClientConnectedEventArgs + var eventArgs = new MqttServerClientConnectedEventArgs { ClientId = connectPacket.ClientId, UserName = connectPacket.Username, ProtocolVersion = channelAdapter.PacketFormatterAdapter.ProtocolVersion, Endpoint = channelAdapter.Endpoint - }).ConfigureAwait(false); + }; + + var handler = ClientConnectedHandler; + if (handler != null) + { + await handler.HandleClientConnectedAsync(eventArgs).ConfigureAwait(false); + } + + await ClientConnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } catch (Exception exception) { @@ -57,18 +69,20 @@ namespace MQTTnet.Server.Internal { try { - var handler = ClientDisconnectedHandler; - if (handler == null) - { - return; - } - - await handler.HandleClientDisconnectedAsync(new MqttServerClientDisconnectedEventArgs + var eventArgs = new MqttServerClientDisconnectedEventArgs { ClientId = clientId, DisconnectType = disconnectType, Endpoint = endpoint - }).ConfigureAwait(false); + }; + + var handler = ClientDisconnectedHandler; + if (handler != null) + { + await handler.HandleClientDisconnectedAsync(eventArgs).ConfigureAwait(false); + } + + await ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } catch (Exception exception) { @@ -80,17 +94,19 @@ namespace MQTTnet.Server.Internal { try { - var handler = ClientSubscribedTopicHandler; - if (handler == null) - { - return; - } - - await handler.HandleClientSubscribedTopicAsync(new MqttServerClientSubscribedTopicEventArgs + var eventArgs = new MqttServerClientSubscribedTopicEventArgs { ClientId = clientId, TopicFilter = topicFilter - }).ConfigureAwait(false); + }; + + var handler = ClientSubscribedTopicHandler; + if (handler != null) + { + await handler.HandleClientSubscribedTopicAsync(eventArgs).ConfigureAwait(false); + } + + await ClientSubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } catch (Exception exception) { @@ -102,17 +118,19 @@ namespace MQTTnet.Server.Internal { try { - var handler = ClientUnsubscribedTopicHandler; - if (handler == null) - { - return; - } - - await handler.HandleClientUnsubscribedTopicAsync(new MqttServerClientUnsubscribedTopicEventArgs + var eventArgs = new MqttServerClientUnsubscribedTopicEventArgs { ClientId = clientId, TopicFilter = topicFilter - }).ConfigureAwait(false); + }; + + var handler = ClientUnsubscribedTopicHandler; + if (handler != null) + { + await handler.HandleClientUnsubscribedTopicAsync(eventArgs).ConfigureAwait(false); + } + + await ClientUnsubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } catch (Exception exception) { @@ -124,13 +142,15 @@ namespace MQTTnet.Server.Internal { try { + var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage, null, null); + var handler = ApplicationMessageReceivedHandler; - if (handler == null) + if (handler != null) { - return; + await handler.HandleApplicationMessageReceivedAsync(eventArgs).ConfigureAwait(false); } - await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage, null, null)).ConfigureAwait(false); + await ApplicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } catch (Exception exception) { diff --git a/Source/MQTTnet/Server/Internal/SubscribeResult.cs b/Source/MQTTnet/Server/Internal/MqttSubscribeResult.cs similarity index 92% rename from Source/MQTTnet/Server/Internal/SubscribeResult.cs rename to Source/MQTTnet/Server/Internal/MqttSubscribeResult.cs index 2a3f070c..1bc09c5e 100644 --- a/Source/MQTTnet/Server/Internal/SubscribeResult.cs +++ b/Source/MQTTnet/Server/Internal/MqttSubscribeResult.cs @@ -3,7 +3,7 @@ using MQTTnet.Protocol; namespace MQTTnet.Server.Internal { - public sealed class SubscribeResult + public sealed class MqttSubscribeResult { public List ReturnCodes { get; } = new List(128); diff --git a/Source/MQTTnet/Server/Internal/Subscription.cs b/Source/MQTTnet/Server/Internal/MqttSubscription.cs similarity index 92% rename from Source/MQTTnet/Server/Internal/Subscription.cs rename to Source/MQTTnet/Server/Internal/MqttSubscription.cs index 9dcd4657..9234ea2b 100644 --- a/Source/MQTTnet/Server/Internal/Subscription.cs +++ b/Source/MQTTnet/Server/Internal/MqttSubscription.cs @@ -2,7 +2,7 @@ namespace MQTTnet.Server.Internal { - public sealed class Subscription + public sealed class MqttSubscription { public string Topic { get; set; } diff --git a/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs b/Source/MQTTnet/Server/Internal/MqttUnsubscribeResult.cs similarity index 85% rename from Source/MQTTnet/Server/Internal/UnsubscribeResult.cs rename to Source/MQTTnet/Server/Internal/MqttUnsubscribeResult.cs index 37d8e002..31e0514d 100644 --- a/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs +++ b/Source/MQTTnet/Server/Internal/MqttUnsubscribeResult.cs @@ -3,7 +3,7 @@ using MQTTnet.Protocol; namespace MQTTnet.Server.Internal { - public sealed class UnsubscribeResult + public sealed class MqttUnsubscribeResult { public List ReasonCodes { get; } = new List(128); diff --git a/Source/MQTTnet/Server/MqttApplicationMessageInterceptorContext.cs b/Source/MQTTnet/Server/MqttApplicationMessageInterceptorContext.cs index 7f943d33..bad2bc33 100644 --- a/Source/MQTTnet/Server/MqttApplicationMessageInterceptorContext.cs +++ b/Source/MQTTnet/Server/MqttApplicationMessageInterceptorContext.cs @@ -1,5 +1,5 @@ -using System; -using System.Collections.Generic; +using System.Collections.Generic; +using System.Threading; namespace MQTTnet.Server { @@ -18,18 +18,21 @@ namespace MQTTnet.Server /// public IDictionary SessionItems { get; internal set; } - [Obsolete("Please use ProcessPublish instead.")] - public bool AcceptPublish - { - get => ProcessPublish; - set => ProcessPublish = value; - } + /// + /// Gets the response which will be sent to the client via the PUBACK etc. packets. + /// + public MqttApplicationMessageResponse Response { get; } = new MqttApplicationMessageResponse(); /// /// Gets or sets whether the publish should be processed internally. /// public bool ProcessPublish { get; set; } = true; - + public bool CloseConnection { get; set; } + + /// + /// Gets the cancellation token which can indicate that the client connection gets down. + /// + public CancellationToken CancellationToken { get; internal set; } } } diff --git a/Source/MQTTnet/Server/MqttApplicationMessageResponse.cs b/Source/MQTTnet/Server/MqttApplicationMessageResponse.cs new file mode 100644 index 00000000..5647bbc7 --- /dev/null +++ b/Source/MQTTnet/Server/MqttApplicationMessageResponse.cs @@ -0,0 +1,14 @@ +using System.Collections.Generic; +using MQTTnet.Packets; + +namespace MQTTnet.Server +{ + public sealed class MqttApplicationMessageResponse + { + public MqttApplicationMessageResponseReasonCode ReasonCode { get; set; } = MqttApplicationMessageResponseReasonCode.Success; + + public string ReasonString { get; set; } + + public List UserProperties { get; } = new List(); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttApplicationMessageResponseReasonCode.cs b/Source/MQTTnet/Server/MqttApplicationMessageResponseReasonCode.cs new file mode 100644 index 00000000..73a2795e --- /dev/null +++ b/Source/MQTTnet/Server/MqttApplicationMessageResponseReasonCode.cs @@ -0,0 +1,17 @@ +namespace MQTTnet.Server +{ + public enum MqttApplicationMessageResponseReasonCode + { + + Success = 0, + //NoMatchingSubscribers = 16, + UnspecifiedError = 128, + ImplementationSpecificError = 131, + NotAuthorized = 135, + TopicNameInvalid = 144, + //PacketIdentifierInUse = 145, + QuotaExceeded = 151, + PayloadFormatInvalid = 153 + + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 48338716..0ae875e4 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -10,16 +10,17 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; using MQTTnet.Diagnostics.Logger; -using MQTTnet.Formatter; using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.Server.Internal; namespace MQTTnet.Server { - public class MqttServer : Disposable, IMqttServer + public sealed class MqttServer : Disposable, IMqttServer { - readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); + readonly AsyncEvent _startedEvent = new AsyncEvent(); + readonly AsyncEvent _stoppedEvent = new AsyncEvent(); + readonly MqttServerEventDispatcher _eventDispatcher; readonly ICollection _adapters; readonly IMqttNetLogger _rootLogger; @@ -46,38 +47,80 @@ namespace MQTTnet.Server public IMqttServerStartedHandler StartedHandler { get; set; } + public event Func StartedAsync + { + add => _startedEvent.AddHandler(value); + remove => _startedEvent.RemoveHandler(value); + } + public IMqttServerStoppedHandler StoppedHandler { get; set; } + public event Func StoppedAsync + { + add => _stoppedEvent.AddHandler(value); + remove => _stoppedEvent.RemoveHandler(value); + } + public IMqttServerClientConnectedHandler ClientConnectedHandler { get => _eventDispatcher.ClientConnectedHandler; set => _eventDispatcher.ClientConnectedHandler = value; } + public event Func ClientConnectedAsync + { + add => _eventDispatcher.ClientConnectedEvent.AddHandler(value); + remove => _eventDispatcher.ClientConnectedEvent.RemoveHandler(value); + } + public IMqttServerClientDisconnectedHandler ClientDisconnectedHandler { get => _eventDispatcher.ClientDisconnectedHandler; set => _eventDispatcher.ClientDisconnectedHandler = value; } + public event Func ClientDisconnectedAsync + { + add => _eventDispatcher.ClientDisconnectedEvent.AddHandler(value); + remove => _eventDispatcher.ClientDisconnectedEvent.RemoveHandler(value); + } + public IMqttServerClientSubscribedTopicHandler ClientSubscribedTopicHandler { get => _eventDispatcher.ClientSubscribedTopicHandler; set => _eventDispatcher.ClientSubscribedTopicHandler = value; } + public event Func ClientSubscribedTopicAsync + { + add => _eventDispatcher.ClientSubscribedTopicEvent.AddHandler(value); + remove => _eventDispatcher.ClientSubscribedTopicEvent.RemoveHandler(value); + } + public IMqttServerClientUnsubscribedTopicHandler ClientUnsubscribedTopicHandler { get => _eventDispatcher.ClientUnsubscribedTopicHandler; set => _eventDispatcher.ClientUnsubscribedTopicHandler = value; } + + public event Func ClientUnsubscribedTopicAsync + { + add => _eventDispatcher.ClientUnsubscribedTopicEvent.AddHandler(value); + remove => _eventDispatcher.ClientUnsubscribedTopicEvent.RemoveHandler(value); + } public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get => _eventDispatcher.ApplicationMessageReceivedHandler; set => _eventDispatcher.ApplicationMessageReceivedHandler = value; } - + + public event Func ApplicationMessageReceivedAsync + { + add => _eventDispatcher.ApplicationMessageReceivedEvent.AddHandler(value); + remove => _eventDispatcher.ApplicationMessageReceivedEvent.RemoveHandler(value); + } + public IMqttServerOptions Options { get; private set; } public Task> GetClientStatusAsync() @@ -139,7 +182,7 @@ namespace MQTTnet.Server return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters); } - public Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) + public async Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); @@ -149,10 +192,9 @@ namespace MQTTnet.Server ThrowIfNotStarted(); - var publishPacket = _packetFactories.Publish.Create(applicationMessage); - _clientSessionsManager.DispatchPublishPacket(publishPacket, null); + await _clientSessionsManager.DispatchPublishPacket(applicationMessage, null); - return Task.FromResult(new MqttClientPublishResult()); + return new MqttClientPublishResult(); } public async Task StartAsync(IMqttServerOptions options) @@ -171,7 +213,7 @@ namespace MQTTnet.Server await _retainedMessagesManager.LoadMessagesAsync().ConfigureAwait(false); _clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _eventDispatcher, _rootLogger); - _clientSessionsManager.Start(cancellationToken); + //_clientSessionsManager.Start(cancellationToken); _keepAliveMonitor = new MqttServerKeepAliveMonitor(Options, _clientSessionsManager, _rootLogger); _keepAliveMonitor.Start(cancellationToken); @@ -189,6 +231,8 @@ namespace MQTTnet.Server { await startedHandler.HandleServerStartedAsync(EventArgs.Empty).ConfigureAwait(false); } + + await _startedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); } public async Task StopAsync() @@ -214,7 +258,7 @@ namespace MQTTnet.Server } finally { - _clientSessionsManager?.Dispose(); + //_clientSessionsManager?.Dispose(); _clientSessionsManager = null; _cancellationTokenSource?.Dispose(); @@ -228,6 +272,8 @@ namespace MQTTnet.Server { await stoppedHandler.HandleServerStoppedAsync(EventArgs.Empty).ConfigureAwait(false); } + + await _stoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); } protected override void Dispose(bool disposing) diff --git a/Source/MQTTnet/Server/MqttSubscribeResponse.cs b/Source/MQTTnet/Server/MqttSubscribeResponse.cs new file mode 100644 index 00000000..d0b31010 --- /dev/null +++ b/Source/MQTTnet/Server/MqttSubscribeResponse.cs @@ -0,0 +1,20 @@ +using System.Collections.Generic; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Server +{ + public sealed class MqttSubscribeResponse + { + /// + /// Gets or sets the reason code which is sent to the client. + /// The subscription is skipped when the value is not GrantedQoS_. + /// MQTTv5 only. + /// + public MqttSubscribeReasonCode ReasonCode { get; set; } + + public List UserProperties { get; } = new List(); + + public string ReasonString { get; set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs b/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs index b7098001..3ec75a10 100644 --- a/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs +++ b/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs @@ -1,7 +1,5 @@ -using System; -using System.Collections.Generic; +using System.Collections.Generic; using System.Threading; -using MQTTnet.Protocol; using MQTTnet.Server.Status; namespace MQTTnet.Server @@ -24,48 +22,14 @@ namespace MQTTnet.Server /// Gets or sets a key/value collection that can be used to share data within the scope of this session. /// public IDictionary SessionItems { get; internal set; } - - // Will be removed together with "AcceptSubscription". It only stores the default value when setting "AcceptSubscription" to true. - internal MqttSubscribeReasonCode DefaultReasonCode { get; set; } - internal MqttSubscribeReturnCode DefaultReturnCode { get; set; } - - [Obsolete("Please use a proper value for 'ReasonCode' instead. This property will be removed in the future.")] - public bool AcceptSubscription - { - get => ReasonCode == MqttSubscribeReasonCode.GrantedQoS0 && ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2; - set - { - if (value) - { - ReasonCode = DefaultReasonCode; - ReturnCode = DefaultReturnCode; - } - else - { - ReturnCode = MqttSubscribeReturnCode.Failure; - ReasonCode = MqttSubscribeReasonCode.UnspecifiedError; - } - } - } - /// /// Gets or sets whether the broker should create an internal subscription for the client. /// The broker can also avoid this and return "success" to the client. /// This feature allows using the MQTT Broker as the Frontend and another system as the backend. /// public bool ProcessSubscription { get; set; } = true; - - /// - /// Gets or sets the reason code which is sent to the client. - /// The subscription is skipped when the value is not GrantedQoS_. - /// MQTTv5 only. - /// - public MqttSubscribeReasonCode ReasonCode { get; set; } - - // MQTT < 5 only! - internal MqttSubscribeReturnCode ReturnCode { get; set; } - + /// /// Gets or sets whether the broker should close the client connection. /// @@ -75,6 +39,11 @@ namespace MQTTnet.Server /// Gets the current client session. /// public IMqttSessionStatus Session { get; internal set; } + + /// + /// Gets the response which will be sent to the client via the SUBACK packet. + /// + public MqttSubscribeResponse Response { get; } = new MqttSubscribeResponse(); /// /// Gets the cancellation token which can indicate that the client connection gets down. diff --git a/Source/MQTTnet/Server/MqttUnsubscribeResponse.cs b/Source/MQTTnet/Server/MqttUnsubscribeResponse.cs new file mode 100644 index 00000000..6735cd6f --- /dev/null +++ b/Source/MQTTnet/Server/MqttUnsubscribeResponse.cs @@ -0,0 +1,19 @@ +using System.Collections.Generic; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Server +{ + public sealed class MqttUnsubscribeResponse + { + /// + /// Gets or sets the reason code which is sent to the client. + /// MQTTv5 only. + /// + public MqttUnsubscribeReasonCode ReasonCode { get; set; } + + public List UserProperties { get; } = new List(); + + public string ReasonString { get; set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs b/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs index 19ac121a..4371f38b 100644 --- a/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs +++ b/Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs @@ -1,7 +1,5 @@ -using System; -using System.Collections.Generic; +using System.Collections.Generic; using System.Threading; -using MQTTnet.Protocol; namespace MQTTnet.Server { @@ -25,21 +23,10 @@ namespace MQTTnet.Server /// public IDictionary SessionItems { get; internal set; } - [Obsolete("Please use ReasonCode instead.")] - public bool AcceptUnsubscription - { - get => ReasonCode == MqttUnsubscribeReasonCode.Success; - set => ReasonCode = DefaultReasonCode; - } - - // Gets removed together with "AcceptUnsubscription". - internal MqttUnsubscribeReasonCode DefaultReasonCode { get; set; } - /// - /// Gets or sets the reason code which will be sent to the client. - /// This value is not transmitted when using MQTT v3.1.1. + /// Gets the response which will be sent to the client via the UNSUBACK pocket. /// - public MqttUnsubscribeReasonCode ReasonCode { get; set; } + public MqttUnsubscribeResponse Response { get; } = new MqttUnsubscribeResponse(); /// /// Gets or sets whether the broker should remove an internal subscription for the client. diff --git a/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs b/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs index a98e98ed..1249d62f 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs +++ b/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs @@ -15,7 +15,7 @@ namespace MQTTnet.AspNetCore.Tests { var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311); - var buffer = serializer.Encode(new MqttPublishPacket() {Topic = "a", Payload = new byte[5]}); + var buffer = serializer.Encode(new MqttPublishPacket {Topic = "a", Payload = new byte[5]}).ToArray(); var sequence = new ReadOnlySequence(buffer.Array, buffer.Offset, buffer.Count); diff --git a/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs b/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs index 34a05a9b..7adef004 100644 --- a/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs @@ -28,7 +28,7 @@ namespace MQTTnet.Benchmarks var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311); - var serializedPacket = Join(serializer.Encode(_packet)); + var serializedPacket = Join(serializer.Encode(_packet).ToArray()); _iterations = 10000; diff --git a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs index e2b2bcd8..45ca2f89 100644 --- a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs @@ -31,7 +31,7 @@ namespace MQTTnet.Benchmarks }; _serializer = new MqttV311PacketFormatter(new MqttPacketWriter()); - _serializedPacket = _serializer.Encode(_packet); + _serializedPacket = _serializer.Encode(_packet).ToArray(); } [Benchmark] diff --git a/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs index 77d7d026..0e222907 100644 --- a/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs @@ -84,7 +84,7 @@ namespace MQTTnet.Tests.Client await lowLevelClient.DisconnectAsync(CancellationToken.None).ConfigureAwait(false); Assert.IsNotNull(receivedPacket); - Assert.AreEqual(MqttSubscribeReturnCode.SuccessMaximumQoS0, receivedPacket.ReturnCodes[0]); + Assert.AreEqual(MqttSubscribeReasonCode.GrantedQoS0, receivedPacket.ReasonCodes[0]); } } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs index a314b9a0..1da5c719 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs @@ -36,18 +36,36 @@ namespace MQTTnet.Tests.Mockups set => Implementation.ConnectedHandler = value; } + public event Func ConnectedAsync + { + add => Implementation.ConnectedAsync += value; + remove => Implementation.ConnectedAsync -= value; + } + public IMqttClientDisconnectedHandler DisconnectedHandler { get => Implementation.DisconnectedHandler; set => Implementation.DisconnectedHandler = value; } + public event Func DisconnectedAsync + { + add => Implementation.DisconnectedAsync += value; + remove => Implementation.DisconnectedAsync -= value; + } + public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get => Implementation.ApplicationMessageReceivedHandler; set => Implementation.ApplicationMessageReceivedHandler = value; } + public event Func ApplicationMessageReceivedAsync + { + add => Implementation.ApplicationMessageReceivedAsync += value; + remove => Implementation.ApplicationMessageReceivedAsync -= value; + } + public Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken) { if (TestContext != null) diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs index 50168a5f..54c76bcb 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs @@ -69,6 +69,12 @@ namespace MQTTnet.Tests.Mockups set => Implementation.ApplicationMessageReceivedHandler = value; } + public event Func ApplicationMessageReceivedAsync + { + add => Implementation.ApplicationMessageReceivedAsync += value; + remove => Implementation.ApplicationMessageReceivedAsync -= value; + } + public Task ClearRetainedApplicationMessagesAsync() { return Implementation.ClearRetainedApplicationMessagesAsync(); diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs index 689fb3c2..a89aa00d 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs @@ -513,10 +513,10 @@ namespace MQTTnet.Tests PacketIdentifier = 123 }; - p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS0); - p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS1); - p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS2); - p.ReturnCodes.Add(MqttSubscribeReturnCode.Failure); + p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS0); + p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS1); + p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS2); + p.ReasonCodes.Add(MqttSubscribeReasonCode.UnspecifiedError); SerializeAndCompare(p, "kAYAewABAoA="); } @@ -529,10 +529,10 @@ namespace MQTTnet.Tests PacketIdentifier = 123 }; - p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS0); - p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS1); - p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS2); - p.ReturnCodes.Add(MqttSubscribeReturnCode.Failure); + p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS0); + p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS1); + p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS2); + p.ReasonCodes.Add(MqttSubscribeReasonCode.UnspecifiedError); DeserializeAndCompare(p, "kAYAewABAoA="); } @@ -596,7 +596,7 @@ namespace MQTTnet.Tests byte[] Serialize(MqttBasePacket packet, MqttProtocolVersion protocolVersion) { - return MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, WriterFactory()).Encode(packet).ToArray(); + return MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, WriterFactory()).Encode(packet).ToArray().ToArray(); } protected virtual IMqttPacketWriter WriterFactory() @@ -616,7 +616,7 @@ namespace MQTTnet.Tests var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer); var buffer1 = serializer.Encode(packet); - using (var headerStream = new MemoryStream(buffer1.ToArray())) + using (var headerStream = new MemoryStream(buffer1.ToArray().ToArray())) { var channel = new TestMqttChannel(headerStream); var adapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(protocolVersion, writer), null, new MqttNetEventLogger()); @@ -624,20 +624,7 @@ namespace MQTTnet.Tests var buffer2 = serializer.Encode(receivedPacket); - Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer2.ToArray())); - - //adapter.ReceivePacketAsync(CancellationToken.None); - //var fixedHeader = new byte[2]; - //var header = new MqttPacketReader(channel).ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult().FixedHeader; - - //using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength)) - //{ - // var reader = ReaderFactory(bodyStream.ToArray()); - // var deserializedPacket = serializer.Decode(new ReceivedMqttPacket(header.Flags, reader, 0)); - // var buffer2 = serializer.Encode(deserializedPacket); - - // Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2))); - //} + Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer2.ToArray().ToArray())); } } @@ -648,26 +635,9 @@ namespace MQTTnet.Tests var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer); var buffer = serializer.Encode(packet); - var channel = new TestMqttChannel(buffer.ToArray()); + var channel = new TestMqttChannel(buffer.ToArray().ToArray()); var adapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(protocolVersion, writer), null, new MqttNetEventLogger()); return (TPacket)adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult(); - - //using (var headerStream = new MemoryStream(buffer1.ToArray())) - //{ - - - - - // //var fixedHeader = new byte[2]; - - // //var header = new MqttPacketReader(channel).ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult().FixedHeader; - - // //using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, (int)header.RemainingLength)) - // //{ - // // var reader = ReaderFactory(bodyStream.ToArray()); - // // return (T)serializer.Decode(new ReceivedMqttPacket(header.Flags, reader, 0)); - // //} - //} } MqttProtocolVersion DeserializeAndDetectVersion(MqttPacketFormatterAdapter packetFormatterAdapter, byte[] buffer) @@ -677,24 +647,6 @@ namespace MQTTnet.Tests adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult(); return packetFormatterAdapter.ProtocolVersion; - - //using (var headerStream = new MemoryStream(buffer)) - //{ - - - - - // //var fixedHeader = new byte[2]; - // //var header = new MqttPacketReader(channel).ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult().FixedHeader; - - // //using (var bodyStream = new MemoryStream(buffer, (int)headerStream.Position, (int)header.RemainingLength)) - // //{ - // // var reader = ReaderFactory(bodyStream.ToArray()); - // // var packet = new ReceivedMqttPacket(header.Flags, reader, 0); - // // packetFormatterAdapter.DetectProtocolVersion(packet); - // // return adapter.ProtocolVersion; - // //} - //} } } } diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index 5ba35b57..0e452af6 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -878,7 +878,7 @@ namespace MQTTnet.Tests.Server // This should lead to no subscriptions for "n" at all. So also no sending of retained messages. if (c.TopicFilter.Topic == "n") { - c.ReasonCode = MqttSubscribeReasonCode.UnspecifiedError; + c.Response.ReasonCode = MqttSubscribeReasonCode.UnspecifiedError; } })); diff --git a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs index 2a59c717..88601e8e 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs @@ -34,7 +34,7 @@ namespace MQTTnet.Tests.Server { if (context.SessionItems["can_subscribe_x"] as bool? == false) { - context.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError; + context.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError; } } }) diff --git a/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs index a6d035cc..16c39446 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs @@ -265,7 +265,7 @@ namespace MQTTnet.Tests.Server { if (c.TopicFilter.Topic == "not_allowed_topic") { - c.ReasonCode = MqttSubscribeReasonCode.TopicFilterInvalid; + c.Response.ReasonCode = MqttSubscribeReasonCode.TopicFilterInvalid; } })); diff --git a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs index e321fd29..64b43b04 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs @@ -73,12 +73,12 @@ namespace MQTTnet.TestApp.NetCore { if (context.TopicFilter.Topic.StartsWith("admin/foo/bar") && context.ClientId != "theAdmin") { - context.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError; + context.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError; } if (context.TopicFilter.Topic.StartsWith("the/secret/stuff") && context.ClientId != "Imperator") { - context.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError; + context.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError; context.CloseConnection = true; } })