diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 2c200893..bb6012d3 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -20,6 +20,11 @@ * [Server] Improved topic filter comparisons (support for $). * [Server] Added more MQTTv5 response information to all interceptors (BREAKING CHANGE!). * [Server] Improved session management for MQTT v5 (#1294, thanks to @logicaloud). +* [Server] All interceptors and events are migrated from interfaces to simple events. All existing APIs are availble but must be migrated to corresponding events. +* [Server] Removed all interceptor and event interfaces including the delegate implementations etc. +* [Server] Renamed a lot of classes and adjsuted namespaces. +* [Server] Introduced a new queueing approach for internal message process (packet bus). + Git commit: $gitCommit Copyright Christian Kratky 2016-2021 diff --git a/MQTTnet.sln.DotSettings b/MQTTnet.sln.DotSettings index bcb734e5..e5f37ff2 100644 --- a/MQTTnet.sln.DotSettings +++ b/MQTTnet.sln.DotSettings @@ -1,9 +1,244 @@  + Required + Required + Required + Required NEVER False False False 180 + <Patterns xmlns="urn:schemas-jetbrains-com:member-reordering-patterns"> + <TypePattern DisplayName="Non-reorderable types"> + <TypePattern.Match> + <Or> + <And> + <Kind Is="Interface" /> + <Or> + <HasAttribute Name="System.Runtime.InteropServices.InterfaceTypeAttribute" /> + <HasAttribute Name="System.Runtime.InteropServices.ComImport" /> + </Or> + </And> + <Kind Is="Struct" /> + <HasAttribute Name="JetBrains.Annotations.NoReorderAttribute" /> + <HasAttribute Name="JetBrains.Annotations.NoReorder" /> + </Or> + </TypePattern.Match> + </TypePattern> + + <TypePattern DisplayName="xUnit.net Test Classes" RemoveRegions="All"> + <TypePattern.Match> + <And> + <Kind Is="Class" /> + <HasMember> + <And> + <Kind Is="Method" /> + <HasAttribute Name="Xunit.FactAttribute" Inherited="True" /> + <HasAttribute Name="Xunit.TheoryAttribute" Inherited="True" /> + </And> + </HasMember> + </And> + </TypePattern.Match> + + <Entry DisplayName="Setup/Teardown Methods"> + <Entry.Match> + <Or> + <Kind Is="Constructor" /> + <And> + <Kind Is="Method" /> + <ImplementsInterface Name="System.IDisposable" /> + </And> + </Or> + </Entry.Match> + + <Entry.SortBy> + <Kind> + <Kind.Order> + <DeclarationKind>Constructor</DeclarationKind> + </Kind.Order> + </Kind> + </Entry.SortBy> + </Entry> + + + <Entry DisplayName="All other members" /> + + <Entry DisplayName="Test Methods" Priority="100"> + <Entry.Match> + <And> + <Kind Is="Method" /> + <HasAttribute Name="Xunit.FactAttribute" Inherited="false" /> + <HasAttribute Name="Xunit.TheoryAttribute" Inherited="false" /> + </And> + </Entry.Match> + + <Entry.SortBy> + <Name /> + </Entry.SortBy> + </Entry> + </TypePattern> + + <TypePattern DisplayName="NUnit Test Fixtures" RemoveRegions="All"> + <TypePattern.Match> + <And> + <Kind Is="Class" /> + <Or> + <HasAttribute Name="NUnit.Framework.TestFixtureAttribute" Inherited="true" /> + <HasAttribute Name="NUnit.Framework.TestFixtureSourceAttribute" Inherited="true" /> + <HasMember> + <And> + <Kind Is="Method" /> + <HasAttribute Name="NUnit.Framework.TestAttribute" Inherited="false" /> + <HasAttribute Name="NUnit.Framework.TestCaseAttribute" Inherited="false" /> + <HasAttribute Name="NUnit.Framework.TestCaseSourceAttribute" Inherited="false" /> + </And> + </HasMember> + </Or> + </And> + </TypePattern.Match> + + <Entry DisplayName="Setup/Teardown Methods"> + <Entry.Match> + <And> + <Kind Is="Method" /> + <Or> + <HasAttribute Name="NUnit.Framework.SetUpAttribute" Inherited="true" /> + <HasAttribute Name="NUnit.Framework.TearDownAttribute" Inherited="true" /> + <HasAttribute Name="NUnit.Framework.TestFixtureSetUpAttribute" Inherited="true" /> + <HasAttribute Name="NUnit.Framework.TestFixtureTearDownAttribute" Inherited="true" /> + <HasAttribute Name="NUnit.Framework.OneTimeSetUpAttribute" Inherited="true" /> + <HasAttribute Name="NUnit.Framework.OneTimeTearDownAttribute" Inherited="true" /> + </Or> + </And> + </Entry.Match> + </Entry> + + <Entry DisplayName="All other members" /> + + <Entry DisplayName="Test Methods" Priority="100"> + <Entry.Match> + <And> + <Kind Is="Method" /> + <HasAttribute Name="NUnit.Framework.TestAttribute" Inherited="false" /> + <HasAttribute Name="NUnit.Framework.TestCaseAttribute" Inherited="false" /> + <HasAttribute Name="NUnit.Framework.TestCaseSourceAttribute" Inherited="false" /> + </And> + </Entry.Match> + + <Entry.SortBy> + <Name /> + </Entry.SortBy> + </Entry> + </TypePattern> + + <TypePattern DisplayName="Default Pattern"> + <Entry DisplayName="Public Delegates" Priority="100"> + <Entry.Match> + <And> + <Access Is="Public" /> + <Kind Is="Delegate" /> + </And> + </Entry.Match> + + <Entry.SortBy> + <Name /> + </Entry.SortBy> + </Entry> + + <Entry DisplayName="Public Enums" Priority="100"> + <Entry.Match> + <And> + <Access Is="Public" /> + <Kind Is="Enum" /> + </And> + </Entry.Match> + + <Entry.SortBy> + <Name /> + </Entry.SortBy> + </Entry> + + <Entry DisplayName="Static Fields and Constants"> + <Entry.Match> + <Or> + <Kind Is="Constant" /> + <And> + <Kind Is="Field" /> + <Static /> + </And> + </Or> + </Entry.Match> + + <Entry.SortBy> + <Kind> + <Kind.Order> + <DeclarationKind>Constant</DeclarationKind> + <DeclarationKind>Field</DeclarationKind> + </Kind.Order> + </Kind> + </Entry.SortBy> + </Entry> + + <Entry DisplayName="Fields"> + <Entry.Match> + <And> + <Kind Is="Field" /> + <Not> + <Static /> + </Not> + </And> + </Entry.Match> + + <Entry.SortBy> + <Readonly /> + <Name /> + </Entry.SortBy> + </Entry> + + <Entry DisplayName="Constructors"> + <Entry.Match> + <Kind Is="Constructor" /> + </Entry.Match> + + <Entry.SortBy> + <Static/> + </Entry.SortBy> + </Entry> + + <Entry DisplayName="Events"> + <Entry.Match> + <Kind Is="Event" /> + </Entry.Match> + + <Entry.SortBy> + <Name Direction="Ascending" /> + </Entry.SortBy> + </Entry> + + <Entry DisplayName="Properties, Indexers"> + <Entry.Match> + <Or> + <Kind Is="Property" /> + <Kind Is="Indexer" /> + </Or> + </Entry.Match> + </Entry> + + <Entry DisplayName="All other members"> + <Entry.SortBy> + <Access Order="public internal protected private" /> + <Name Direction="Ascending" /> + </Entry.SortBy> + </Entry> + + <Entry DisplayName="Nested Types"> + <Entry.Match> + <Kind Is="Type" /> + </Entry.Match> + </Entry> + </TypePattern> +</Patterns> + True True True diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 389ec72e..71341047 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -10,6 +10,7 @@ using System.Threading; using System.Threading.Tasks; using MQTTnet.Diagnostics; using MQTTnet.Packets; +using IMqttClient = MQTTnet.Client.IMqttClient; namespace MQTTnet.Extensions.ManagedClient { diff --git a/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs index 043357fa..85438cab 100644 --- a/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs @@ -6,14 +6,14 @@ namespace MQTTnet.Formatter { public sealed class MqttConnAckPacketFactory { - public MqttConnAckPacket Create(ValidatingConnectionEventArgs clientConnectionValidationEventArgs) + public MqttConnAckPacket Create(ValidatingConnectionEventArgs validatingConnectionEventArgs) { - if (clientConnectionValidationEventArgs == null) throw new ArgumentNullException(nameof(clientConnectionValidationEventArgs)); + if (validatingConnectionEventArgs == null) throw new ArgumentNullException(nameof(validatingConnectionEventArgs)); var connAckPacket = new MqttConnAckPacket { - ReturnCode = MqttConnectReasonCodeConverter.ToConnectReturnCode(clientConnectionValidationEventArgs.ReasonCode), - ReasonCode = clientConnectionValidationEventArgs.ReasonCode, + ReturnCode = MqttConnectReasonCodeConverter.ToConnectReturnCode(validatingConnectionEventArgs.ReasonCode), + ReasonCode = validatingConnectionEventArgs.ReasonCode, Properties = { RetainAvailable = true, @@ -22,17 +22,17 @@ namespace MQTTnet.Formatter TopicAliasMaximum = ushort.MaxValue, WildcardSubscriptionAvailable = true, - AuthenticationMethod = clientConnectionValidationEventArgs.AuthenticationMethod, - AuthenticationData = clientConnectionValidationEventArgs.ResponseAuthenticationData, - AssignedClientIdentifier = clientConnectionValidationEventArgs.AssignedClientIdentifier, - ReasonString = clientConnectionValidationEventArgs.ReasonString, - ServerReference = clientConnectionValidationEventArgs.ServerReference + AuthenticationMethod = validatingConnectionEventArgs.AuthenticationMethod, + AuthenticationData = validatingConnectionEventArgs.ResponseAuthenticationData, + AssignedClientIdentifier = validatingConnectionEventArgs.AssignedClientIdentifier, + ReasonString = validatingConnectionEventArgs.ReasonString, + ServerReference = validatingConnectionEventArgs.ServerReference } }; - if (clientConnectionValidationEventArgs.ResponseUserProperties != null) + if (validatingConnectionEventArgs.ResponseUserProperties != null) { - connAckPacket.Properties.UserProperties.AddRange(clientConnectionValidationEventArgs.ResponseUserProperties); + connAckPacket.Properties.UserProperties.AddRange(validatingConnectionEventArgs.ResponseUserProperties); } return connAckPacket; diff --git a/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs b/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs index 82f36f1c..f8891342 100644 --- a/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs @@ -6,6 +6,14 @@ namespace MQTTnet.Formatter { public sealed class MqttDisconnectPacketFactory { + public MqttDisconnectPacket Create(MqttDisconnectReasonCode reasonCode) + { + return new MqttDisconnectPacket + { + ReasonCode = reasonCode + }; + } + public MqttDisconnectPacket Create(MqttClientDisconnectOptions clientDisconnectOptions) { var packet = new MqttDisconnectPacket(); diff --git a/Source/MQTTnet/Internal/AsyncEvent.cs b/Source/MQTTnet/Internal/AsyncEvent.cs index 9fbaa19c..9a60c505 100644 --- a/Source/MQTTnet/Internal/AsyncEvent.cs +++ b/Source/MQTTnet/Internal/AsyncEvent.cs @@ -2,39 +2,49 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; +using MQTTnet.Diagnostics; namespace MQTTnet.Internal { public sealed class AsyncEvent where TEventArgs : EventArgs { - readonly List> _handlers = new List>(); + readonly List> _handlers = new List>(); public void AddHandler(Func handler) { - if (handler == null) throw new ArgumentNullException(nameof(handler)); - - _handlers.Add(handler); - } + if (handler == null) + { + throw new ArgumentNullException(nameof(handler)); + } - public void RemoveHandler(Func handler) + _handlers.Add(new AsyncEventInvocator(null, handler)); + } + + public void AddHandler(Action handler) { - if (handler == null) throw new ArgumentNullException(nameof(handler)); - - _handlers.Remove(handler); + if (handler == null) + { + throw new ArgumentNullException(nameof(handler)); + } + + _handlers.Add(new AsyncEventInvocator(handler, null)); } public async Task InvokeAsync(TEventArgs eventArgs) { foreach (var handler in _handlers) { - await handler.Invoke(eventArgs); + await handler.InvokeAsync(eventArgs).ConfigureAwait(false); } } public async Task InvokeAsync(Func eventArgsProvider) { - if (eventArgsProvider == null) throw new ArgumentNullException(nameof(eventArgsProvider)); - + if (eventArgsProvider == null) + { + throw new ArgumentNullException(nameof(eventArgsProvider)); + } + if (!_handlers.Any()) { return default; @@ -44,5 +54,47 @@ namespace MQTTnet.Internal await InvokeAsync(eventArgs).ConfigureAwait(false); return eventArgs; } + + public void RemoveHandler(Func handler) + { + if (handler == null) + { + throw new ArgumentNullException(nameof(handler)); + } + + _handlers.RemoveAll(h => h.WrapsHandler(handler)); + } + + public void RemoveHandler(Action handler) + { + if (handler == null) + { + throw new ArgumentNullException(nameof(handler)); + } + + _handlers.RemoveAll(h => h.WrapsHandler(handler)); + } + + public async Task TryInvokeAsync(TEventArgs eventArgs, MqttNetSourceLogger logger) + { + if (eventArgs == null) + { + throw new ArgumentNullException(nameof(eventArgs)); + } + + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } + + try + { + await InvokeAsync(eventArgs).ConfigureAwait(false); + } + catch (Exception exception) + { + logger.Warning(exception, $"Error while invoking event ({typeof(TEventArgs)})."); + } + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Internal/AsyncEventInvocator.cs b/Source/MQTTnet/Internal/AsyncEventInvocator.cs new file mode 100644 index 00000000..d956dd9d --- /dev/null +++ b/Source/MQTTnet/Internal/AsyncEventInvocator.cs @@ -0,0 +1,44 @@ +using System; +using System.Threading.Tasks; +using MQTTnet.Implementations; + +namespace MQTTnet.Internal +{ + public sealed class AsyncEventInvocator + { + readonly Action _handler; + readonly Func _asyncHandler; + + public AsyncEventInvocator(Action handler, Func asyncHandler) + { + _handler = handler; + _asyncHandler = asyncHandler; + } + + public bool WrapsHandler(Action handler) + { + return ReferenceEquals(_handler, handler); + } + + public bool WrapsHandler(Func handler) + { + return ReferenceEquals(_asyncHandler, handler); + } + + public Task InvokeAsync(TEventArgs eventArgs) + { + if (_handler != null) + { + _handler.Invoke(eventArgs); + return PlatformAbstractionLayer.CompletedTask; + } + + if (_asyncHandler != null) + { + return _asyncHandler.Invoke(eventArgs); + } + + throw new InvalidOperationException(); + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Internal/MqttPacketBus.cs b/Source/MQTTnet/Internal/MqttPacketBus.cs index 446c26a6..0d7e8132 100644 --- a/Source/MQTTnet/Internal/MqttPacketBus.cs +++ b/Source/MQTTnet/Internal/MqttPacketBus.cs @@ -9,9 +9,7 @@ namespace MQTTnet.Internal { public sealed class MqttPacketBus : IDisposable { - readonly object _syncRoot = new object(); - - readonly LinkedList[] _partitions = new LinkedList[] + readonly LinkedList[] _partitions = { new LinkedList(), new LinkedList(), @@ -19,10 +17,11 @@ namespace MQTTnet.Internal }; readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0); + readonly object _syncRoot = new object(); int _activePartition = (int) MqttPacketBusPartition.Health; - public int PacketsCount + public int ItemsCount { get { @@ -33,50 +32,18 @@ namespace MQTTnet.Internal } } - public int DataPacketsCount + public void Clear() { - get + lock (_syncRoot) { - lock (_syncRoot) + foreach (var partition in _partitions) { - return _partitions[(int) MqttPacketBusPartition.Data].Count; + partition.Clear(); } } } - public List ExportPackets(MqttPacketBusPartition partition) - { - lock (_syncRoot) - { - return _partitions[(int) partition].Select(i => i.Packet).ToList(); - } - } - - public void Enqueue(MqttPacketBusItem item, MqttPacketBusPartition partition) - { - if (item == null) throw new ArgumentNullException(nameof(item)); - - lock (_syncRoot) - { - _partitions[(int) partition].AddLast(item); - } - - _semaphore.Release(); - } - - void MoveActivePartition() - { - if (_activePartition >= _partitions.Length - 1) - { - _activePartition = 0; - } - else - { - _activePartition++; - } - } - - public async Task DequeueAsync(CancellationToken cancellationToken) + public async Task DequeueItemAsync(CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) { @@ -99,7 +66,7 @@ namespace MQTTnet.Internal // the worker back to the thread pool. await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); } - + cancellationToken.ThrowIfCancellationRequested(); throw new InvalidOperationException("MqttPacketBus is broken."); @@ -110,15 +77,60 @@ namespace MQTTnet.Internal _semaphore?.Dispose(); } - public void Clear() + public void DropFirstItem(MqttPacketBusPartition partition) { lock (_syncRoot) { - foreach (var partition in _partitions) + var partitionInstance = _partitions[(int) partition]; + + if (partitionInstance.Any()) { - partition.Clear(); + partitionInstance.RemoveFirst(); } } } + + public void EnqueueItem(MqttPacketBusItem item, MqttPacketBusPartition partition) + { + if (item == null) + { + throw new ArgumentNullException(nameof(item)); + } + + lock (_syncRoot) + { + _partitions[(int) partition].AddLast(item); + } + + _semaphore.Release(); + } + + public List ExportPackets(MqttPacketBusPartition partition) + { + lock (_syncRoot) + { + return _partitions[(int) partition].Select(i => i.Packet).ToList(); + } + } + + void MoveActivePartition() + { + if (_activePartition >= _partitions.Length - 1) + { + _activePartition = 0; + } + else + { + _activePartition++; + } + } + + public int PartitionItemsCount(MqttPacketBusPartition partition) + { + lock (_syncRoot) + { + return _partitions[(int) partition].Count; + } + } } } \ No newline at end of file diff --git a/Source/MQTTnet/MQTTnet.csproj.DotSettings b/Source/MQTTnet/MQTTnet.csproj.DotSettings index da48a98e..2cabc058 100644 --- a/Source/MQTTnet/MQTTnet.csproj.DotSettings +++ b/Source/MQTTnet/MQTTnet.csproj.DotSettings @@ -12,6 +12,7 @@ True True True + True True True True \ No newline at end of file diff --git a/Source/MQTTnet/MqttFactory.cs b/Source/MQTTnet/MqttFactory.cs index 378a864e..34e1e5f4 100644 --- a/Source/MQTTnet/MqttFactory.cs +++ b/Source/MQTTnet/MqttFactory.cs @@ -7,6 +7,7 @@ using System; using System.Collections.Generic; using System.Linq; using MQTTnet.Diagnostics; +using IMqttClient = MQTTnet.Client.IMqttClient; using MqttClient = MQTTnet.Client.MqttClient; namespace MQTTnet diff --git a/Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs index 5054bf39..806a3b34 100644 --- a/Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs +++ b/Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs @@ -1,4 +1,5 @@ using System; +using System.Collections; using System.Collections.Generic; using System.Threading; @@ -17,7 +18,7 @@ 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; } + public IDictionary SessionItems { get; internal set; } /// /// Gets the response which will be sent to the client via the PUBACK etc. packets. diff --git a/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs index 724f2c7e..8cb39fae 100644 --- a/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs @@ -1,5 +1,5 @@ using System; -using System.Collections.Generic; +using System.Collections; using System.Threading; using MQTTnet.Packets; @@ -22,7 +22,7 @@ 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; } + public IDictionary SessionItems { get; internal set; } /// /// Gets or sets whether the broker should create an internal subscription for the client. @@ -39,7 +39,7 @@ namespace MQTTnet.Server /// /// Gets the current client session. /// - public IMqttSessionStatus Session { get; internal set; } + public MqttSessionStatus Session { get; internal set; } /// /// Gets the response which will be sent to the client via the SUBACK packet. diff --git a/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs index 8f90df43..cadb3d4b 100644 --- a/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs @@ -1,4 +1,5 @@ using System; +using System.Collections; using System.Collections.Generic; using System.Threading; @@ -22,7 +23,7 @@ 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; } + public IDictionary SessionItems { get; internal set; } /// /// Gets the response which will be sent to the client via the UNSUBACK pocket. diff --git a/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs b/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs index d557f08e..2e5d3d4d 100644 --- a/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs +++ b/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs @@ -5,6 +5,6 @@ namespace MQTTnet.Server { public sealed class LoadingRetainedMessagesEventArgs : EventArgs { - public List StoredRetainedMessages { get; set; } = new List(); + public List LoadedRetainedMessages { get; set; } = new List(); } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs b/Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs index befa334b..01774bf2 100644 --- a/Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs @@ -26,7 +26,7 @@ namespace MQTTnet.Server public List PublishPackets { get; } = new List(); - IDictionary Items { get; set; } + public IDictionary Items { get; set; } /// /// Gets the last will message. diff --git a/Source/MQTTnet/Server/Events/SessionDeletedEventArgs.cs b/Source/MQTTnet/Server/Events/SessionDeletedEventArgs.cs new file mode 100644 index 00000000..ccee7f50 --- /dev/null +++ b/Source/MQTTnet/Server/Events/SessionDeletedEventArgs.cs @@ -0,0 +1,12 @@ +using System; + +namespace MQTTnet.Server +{ + public sealed class SessionDeletedEventArgs : EventArgs + { + /// + /// Gets the ID of the session. + /// + public string Id { get; internal set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs index 757b6eb2..f19fb542 100644 --- a/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs @@ -1,4 +1,5 @@ using System; +using System.Collections; using System.Collections.Generic; using System.Security.Cryptography.X509Certificates; using System.Text; @@ -122,7 +123,7 @@ 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; } + public IDictionary SessionItems { get; internal set; } /// /// This is used for MQTTv3 only. diff --git a/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs b/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs index dad6ad84..c5114886 100644 --- a/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs +++ b/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs @@ -1,7 +1,8 @@ using System; -using System.Collections.Generic; +using System.Collections; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Protocol; namespace MQTTnet.Server { @@ -9,9 +10,9 @@ namespace MQTTnet.Server { readonly MqttServerEventContainer _eventContainer; readonly string _clientId; - readonly IDictionary _sessionItems; + readonly IDictionary _sessionItems; - public MqttApplicationMessageInterceptorInvoker(MqttServerEventContainer eventContainer, string clientId, IDictionary sessionItems) + public MqttApplicationMessageInterceptorInvoker(MqttServerEventContainer eventContainer, string clientId, IDictionary sessionItems) { _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer)); _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); @@ -49,7 +50,7 @@ namespace MQTTnet.Server { // This can happen if a topic alias us used but the topic is // unknown to the server. - eventArgs.Response.ReasonCode = MqttApplicationMessageResponseReasonCode.TopicNameInvalid; + eventArgs.Response.ReasonCode = MqttPubAckReasonCode.TopicNameInvalid; eventArgs.ProcessPublish = false; } diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 4234d6f5..9c7c314c 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -75,11 +75,11 @@ namespace MQTTnet.Server public bool IsCleanDisconnect { get; private set; } - public async Task StopAsync(MqttClientDisconnectReason reason) + public async Task StopAsync(MqttDisconnectReasonCode reason) { IsRunning = false; - if (reason == MqttClientDisconnectReason.SessionTakenOver || reason == MqttClientDisconnectReason.KeepAliveTimeout) + if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout) { // Is is very important to send the DISCONNECT packet here BEFORE cancelling the // token because the entire connection is closed (disposed) as soon as the cancellation @@ -391,7 +391,7 @@ namespace MQTTnet.Server if (_applicationMessageInterceptorInvoker.CloseConnection) { - await StopAsync(MqttClientDisconnectReason.UnspecifiedError); + await StopAsync(MqttDisconnectReasonCode.UnspecifiedError); return; } @@ -455,18 +455,12 @@ namespace MQTTnet.Server } } - async Task TrySendDisconnectPacket(MqttClientDisconnectReason reason) + async Task TrySendDisconnectPacket(MqttDisconnectReasonCode reasonCode) { try { - var disconnectOptions = new MqttClientDisconnectOptions - { - Reason = reason, - ReasonString = reason.ToString() - }; - - var disconnectPacket = _packetFactories.Disconnect.Create(disconnectOptions); - + var disconnectPacket = _packetFactories.Disconnect.Create(reasonCode); + using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout)) { await SendPacketAsync(disconnectPacket, timeout.Token).ConfigureAwait(false); @@ -474,7 +468,7 @@ namespace MQTTnet.Server } catch (Exception exception) { - _logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (Reason = {1}).", Id, reason); + _logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1}).", Id, reasonCode); } } } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 405bd652..88a198ba 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -1,4 +1,5 @@ using System; +using System.Collections; using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; @@ -22,11 +23,7 @@ namespace MQTTnet.Server readonly AsyncLock _createConnectionSyncRoot = new AsyncLock(); readonly Dictionary _clients = new Dictionary(4096); readonly Dictionary _sessions = new Dictionary(4096); - - readonly IDictionary _serverSessionItems = new ConcurrentDictionary(); - - readonly MqttConnAckPacketFactory _connAckPacketFactory = new MqttConnAckPacketFactory(); - + readonly MqttRetainedMessagesManager _retainedMessagesManager; readonly MqttServerEventContainer _eventContainer; readonly MqttServerOptions _options; @@ -89,21 +86,18 @@ namespace MQTTnet.Server return; } - MqttConnAckPacket connAckPacket; + var validatingConnectionEventArgs = await ValidateConnection(connectPacket, channelAdapter).ConfigureAwait(false); + var connAckPacket = _packetFactories.ConnAck.Create(validatingConnectionEventArgs); - var connectionValidatorContext = await ValidateConnection(connectPacket, channelAdapter).ConfigureAwait(false); - if (connectionValidatorContext.ReasonCode != MqttConnectReasonCode.Success) + if (validatingConnectionEventArgs.ReasonCode != MqttConnectReasonCode.Success) { // Send failure response here without preparing a connection and session! - connAckPacket = _connAckPacketFactory.Create(connectionValidatorContext); await channelAdapter.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false); return; } - connAckPacket = _connAckPacketFactory.Create(connectionValidatorContext); - - // Pass connAckPacket so that IsSessionPresent flag can be set if the client session already exists - client = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter, connectionValidatorContext).ConfigureAwait(false); + // Pass connAckPacket so that IsSessionPresent flag can be set if the client session already exists. + client = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter, validatingConnectionEventArgs).ConfigureAwait(false); await client.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false); @@ -176,21 +170,34 @@ namespace MQTTnet.Server foreach (var connection in connections) { - await connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); + await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false); } } - public List GetConnections() + public List GetClients() { lock (_clients) { return _clients.Values.ToList(); } } - - public Task> GetClientStatusAsync() + + public MqttClient GetClient(string id) { - var result = new List(); + lock (_clients) + { + if (!_clients.TryGetValue(id, out var client)) + { + throw new InvalidOperationException($"Client with ID '{id}' not found."); + } + + return client; + } + } + + public Task> GetClientStatusAsync() + { + var result = new List(); lock (_clients) { @@ -205,12 +212,12 @@ namespace MQTTnet.Server } } - return Task.FromResult((IList)result); + return Task.FromResult((IList)result); } - public Task> GetSessionStatusAsync() + public Task> GetSessionStatusAsync() { - var result = new List(); + var result = new List(); lock (_sessions) { @@ -221,15 +228,8 @@ namespace MQTTnet.Server } } - return Task.FromResult((IList)result); + 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 async Task SubscribeAsync(string clientId, ICollection topicFilters) { @@ -281,7 +281,7 @@ namespace MQTTnet.Server { if (connection != null) { - await connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); + await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false); } } catch (Exception exception) @@ -291,7 +291,10 @@ namespace MQTTnet.Server try { - session?.OnDeleted(); + await _eventContainer.SessionDeletedEvent.TryInvokeAsync(new SessionDeletedEventArgs + { + Id = session.Id + }, _logger).ConfigureAwait(false); } catch (Exception exception) { @@ -302,31 +305,7 @@ namespace MQTTnet.Server _logger.Verbose("Session for client '{0}' deleted.", clientId); } - - // 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."); - // } - // } - // } - + public async Task DispatchPublishPacket(string senderClientId, MqttApplicationMessage applicationMessage) { try @@ -501,7 +480,7 @@ namespace MQTTnet.Server if (existing != null) { existing.IsTakenOver = true; - await existing.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false); + await existing.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false); await _eventContainer.ClientDisconnectedEvent.InvokeAsync(() => new ClientDisconnectedEventArgs { @@ -514,40 +493,7 @@ namespace MQTTnet.Server return connection; } - - // async Task InterceptApplicationMessageAsync( - // IMqttServerApplicationMessageInterceptor interceptor, - // MqttClientConnection clientConnection, - // MqttApplicationMessage applicationMessage) - // { - // string senderClientId; - // IDictionary sessionItems; - // - // var messageIsFromServer = clientConnection == null; - // if (messageIsFromServer) - // { - // senderClientId = _options.ClientId; - // sessionItems = _serverSessionItems; - // } - // else - // { - // senderClientId = clientConnection.Id; - // sessionItems = clientConnection.Session.Items; - // } - // - // var interceptorContext = new InterceptingMqttClientPublishEventArgs - // { - // ClientId = senderClientId, - // SessionItems = sessionItems, - // ProcessPublish = true, - // ApplicationMessage = applicationMessage, - // CloseConnection = false - // }; - // - // await interceptor.InterceptApplicationMessagePublishAsync(interceptorContext).ConfigureAwait(false); - // return interceptorContext; - // } - + MqttSession GetClientSession(string clientId) { lock (_sessions) @@ -573,7 +519,7 @@ namespace MQTTnet.Server _rootLogger); } - MqttSession CreateSession(string clientId, IDictionary sessionItems, bool isPersistent) + MqttSession CreateSession(string clientId, IDictionary sessionItems, bool isPersistent) { _logger.Verbose("Created a new session for client '{0}'.", clientId); diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 5b6e4179..09db68b7 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -13,22 +14,22 @@ namespace MQTTnet.Server // We use a reader writer lock because the subscriptions are only read most of the time. // 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 SemaphoreSlim _subscriptionsLock = new SemaphoreSlim(1); + + // The subscriptions are stored as a ConcurrentDictionary in order ensure that reading the data is save. + // The additional lock is important to coordinate complex update logic with multiple steps, checks and interceptors. + readonly ConcurrentDictionary _subscriptions = new ConcurrentDictionary(); readonly MqttSession _session; - readonly MqttServerOptions _options; readonly MqttServerEventContainer _eventContainer; readonly MqttRetainedMessagesManager _retainedMessagesManager; public MqttClientSubscriptionsManager( MqttSession session, - MqttServerOptions serverOptions, MqttServerEventContainer eventContainer, MqttRetainedMessagesManager retainedMessagesManager) { _session = session ?? throw new ArgumentNullException(nameof(session)); - _options = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer)); _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager)); } @@ -98,7 +99,7 @@ namespace MQTTnet.Server var result = new MqttUnsubscribeResult(); - _subscriptionsLock.EnterWriteLock(); + await _subscriptionsLock.WaitAsync(cancellationToken).ConfigureAwait(false); try { foreach (var topicFilter in unsubscribePacket.TopicFilters) @@ -124,13 +125,13 @@ namespace MQTTnet.Server if (interceptorContext.ProcessUnsubscription) { - _subscriptions.Remove(topicFilter); + _subscriptions.TryRemove(topicFilter, out _); } } } finally { - _subscriptionsLock.ExitWriteLock(); + _subscriptionsLock.Release(); } foreach (var topicFilter in unsubscribePacket.TopicFilters) @@ -148,17 +149,7 @@ namespace MQTTnet.Server public CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel qosLevel, string senderClientId) { - List subscriptions; - _subscriptionsLock.EnterReadLock(); - try - { - subscriptions = _subscriptions.Values.ToList(); - } - finally - { - _subscriptionsLock.ExitReadLock(); - } - + var subscriptions = _subscriptions.Values.ToList(); var senderIsReceiver = string.Equals(senderClientId, _session.Id); var qosLevels = new HashSet(); @@ -243,7 +234,7 @@ namespace MQTTnet.Server bool isNewSubscription; - _subscriptionsLock.EnterWriteLock(); + _subscriptionsLock.Wait(); try { isNewSubscription = !_subscriptions.ContainsKey(topicFilter.Topic); @@ -251,7 +242,7 @@ namespace MQTTnet.Server } finally { - _subscriptionsLock.ExitWriteLock(); + _subscriptionsLock.Release(); } return new CreateSubscriptionResult diff --git a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs index 6764ecb5..f68551dc 100644 --- a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs @@ -36,7 +36,7 @@ namespace MQTTnet.Server { _messages.Clear(); - foreach (var retainedMessage in eventArgs.StoredRetainedMessages ?? new List()) + foreach (var retainedMessage in eventArgs.LoadedRetainedMessages ?? new List()) { _messages[retainedMessage.Topic] = retainedMessage; } diff --git a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs index dbd2cae8..5e53a60a 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs @@ -23,6 +23,8 @@ namespace MQTTnet.Server public AsyncEvent PreparingSessionEvent { get; } = new AsyncEvent(); + public AsyncEvent SessionDeletedEvent { get; } = new AsyncEvent(); + public AsyncEvent ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent(); public AsyncEvent RetainedMessageChangedEvent { get; } = new AsyncEvent(); diff --git a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs index e164d429..533e215a 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs @@ -1,10 +1,10 @@ using System; using System.Threading; using System.Threading.Tasks; -using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Implementations; using MQTTnet.Internal; +using MQTTnet.Protocol; namespace MQTTnet.Server { @@ -40,7 +40,7 @@ namespace MQTTnet.Server while (!cancellationToken.IsCancellationRequested) { - TryMaintainConnections(); + TryProcessClients(); PlatformAbstractionLayer.Sleep(_options.KeepAliveMonitorInterval); } } @@ -57,16 +57,16 @@ namespace MQTTnet.Server } } - void TryMaintainConnections() + void TryProcessClients() { var now = DateTime.UtcNow; - foreach (var connection in _sessionsManager.GetConnections()) + foreach (var client in _sessionsManager.GetClients()) { - TryMaintainConnection(connection, now); + TryProcessClient(client, now); } } - void TryMaintainConnection(MqttClient connection, DateTime now) + void TryProcessClient(MqttClient connection, DateTime now) { try { @@ -108,7 +108,7 @@ namespace MQTTnet.Server // We do not need to wait for the task so no await is needed. // Also the internal state of the connection must be swapped to "Finalizing" because the // next iteration of the keep alive timer happens. - var _ = connection.StopAsync(MqttClientDisconnectReason.KeepAliveTimeout); + var _ = connection.StopAsync(MqttDisconnectReasonCode.KeepAliveTimeout); } catch (Exception exception) { diff --git a/Source/MQTTnet/Server/Internal/MqttSession.cs b/Source/MQTTnet/Server/Internal/MqttSession.cs index 82142b70..4e35680e 100644 --- a/Source/MQTTnet/Server/Internal/MqttSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttSession.cs @@ -1,4 +1,5 @@ using System; +using System.Collections; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -21,7 +22,7 @@ namespace MQTTnet.Server public MqttSession(string clientId, bool isPersistent, - IDictionary items, + IDictionary items, MqttServerOptions serverOptions, MqttServerEventContainer eventContainer, MqttRetainedMessagesManager retainedMessagesManager, @@ -34,11 +35,9 @@ namespace MQTTnet.Server _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager)); - SubscriptionsManager = new MqttClientSubscriptionsManager(this, serverOptions, eventContainer, retainedMessagesManager); + SubscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager); } - - public event EventHandler Deleted; - + public string Id { get; } /// @@ -54,11 +53,11 @@ namespace MQTTnet.Server public MqttClientSubscriptionsManager SubscriptionsManager { get; } - public IDictionary Items { get; } + public IDictionary Items { get; } public bool WillMessageSent { get; set; } - public long PendingDataPacketsCount => _packetBus.DataPacketsCount; + public long PendingDataPacketsCount => _packetBus.PartitionItemsCount(MqttPacketBusPartition.Data); public void AcknowledgePublishPacket(ushort packetIdentifier) { @@ -69,14 +68,16 @@ namespace MQTTnet.Server { if (packetBusItem == null) throw new ArgumentNullException(nameof(packetBusItem)); - if (_packetBus.PacketsCount >= _serverOptions.MaxPendingMessagesPerClient) + if (_packetBus.ItemsCount >= _serverOptions.MaxPendingMessagesPerClient) { if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage) { return; } - else + if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage) { + _packetBus.DropFirstItem(MqttPacketBusPartition.Data); + // TODO: Implement. } } @@ -88,33 +89,28 @@ namespace MQTTnet.Server _unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket; } - _packetBus.Enqueue(packetBusItem, MqttPacketBusPartition.Data); + _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data); } else if (packetBusItem.Packet is MqttPingReqPacket || packetBusItem.Packet is MqttPingRespPacket) { - _packetBus.Enqueue(packetBusItem, MqttPacketBusPartition.Health); + _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health); } else { - _packetBus.Enqueue(packetBusItem, MqttPacketBusPartition.Control); + _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control); } } public Task DequeuePacketAsync(CancellationToken cancellationToken) { - return _packetBus.DequeueAsync(cancellationToken); + return _packetBus.DequeueItemAsync(cancellationToken); } public Task DeleteAsync() { return _clientSessionsManager.DeleteSessionAsync(Id); } - - public void OnDeleted() - { - Deleted?.Invoke(this, EventArgs.Empty); - } - + public void Recover() { // TODO: Keep the bus and only insert pending items again. diff --git a/Source/MQTTnet/Server/MqttApplicationMessageResponseReasonCode.cs b/Source/MQTTnet/Server/MqttApplicationMessageResponseReasonCode.cs deleted file mode 100644 index 1a7073d5..00000000 --- a/Source/MQTTnet/Server/MqttApplicationMessageResponseReasonCode.cs +++ /dev/null @@ -1,16 +0,0 @@ -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/MqttClientMessageQueueInterceptorContext.cs b/Source/MQTTnet/Server/MqttClientMessageQueueInterceptorContext.cs deleted file mode 100644 index 39b08ff7..00000000 --- a/Source/MQTTnet/Server/MqttClientMessageQueueInterceptorContext.cs +++ /dev/null @@ -1,25 +0,0 @@ -using MQTTnet.Protocol; - -namespace MQTTnet.Server -{ - public sealed class MqttClientMessageQueueInterceptorContext - { - public string SenderClientId { get; internal set; } - - public string ReceiverClientId { get; internal set; } - - public MqttApplicationMessage ApplicationMessage { get; set; } - - public bool AcceptEnqueue { get; set; } = true; - - /// - /// Gets or sets the supscription quality of service level. - /// The Quality of Service (QoS) level is an agreement between the sender of a message and the receiver of a message that defines the guarantee of delivery for a specific message. - /// There are 3 QoS levels in MQTT: - /// - At most once (0): Message gets delivered no time, once or multiple times. - /// - At least once (1): Message gets delivered at least once (one time or more often). - /// - Exactly once (2): Message gets delivered exactly once (It's ensured that the message only comes once). - /// - public MqttQualityOfServiceLevel SubscriptionQualityOfServiceLevel { get; set; } - } -} diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index c485a607..ac5daa2f 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -1,120 +1,43 @@ -using MQTTnet.Adapter; -using MQTTnet.Protocol; -using System; +using System; using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; -using MQTTnet.Client; +using MQTTnet.Adapter; using MQTTnet.Diagnostics; using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.Packets; +using MQTTnet.Protocol; namespace MQTTnet.Server { public class MqttServer : Disposable { - readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer(); - readonly ICollection _adapters; + readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer(); + readonly MqttNetSourceLogger _logger; readonly MqttServerOptions _options; readonly IMqttNetLogger _rootLogger; - readonly MqttNetSourceLogger _logger; + CancellationTokenSource _cancellationTokenSource; MqttClientSessionsManager _clientSessionsManager; - MqttRetainedMessagesManager _retainedMessagesManager; MqttServerKeepAliveMonitor _keepAliveMonitor; - CancellationTokenSource _cancellationTokenSource; + MqttRetainedMessagesManager _retainedMessagesManager; public MqttServer(MqttServerOptions options, IEnumerable adapters, IMqttNetLogger logger) { _options = options ?? throw new ArgumentNullException(nameof(options)); - - if (adapters == null) throw new ArgumentNullException(nameof(adapters)); + + if (adapters == null) + { + throw new ArgumentNullException(nameof(adapters)); + } + _adapters = adapters.ToList(); - if (logger == null) throw new ArgumentNullException(nameof(logger)); + _rootLogger = logger ?? throw new ArgumentNullException(nameof(logger)); _logger = logger.WithSource(nameof(MqttServer)); - _rootLogger = logger; - } - - public event Func StartedAsync - { - add => _eventContainer.StartedEvent.AddHandler(value); - remove => _eventContainer.StartedEvent.RemoveHandler(value); - } - - public event Func StoppedAsync - { - add => _eventContainer.StoppedEvent.AddHandler(value); - remove => _eventContainer.StoppedEvent.RemoveHandler(value); - } - - public event Func ValidatingConnectionAsync - { - add => _eventContainer.ValidatingConnectionEvent.AddHandler(value); - remove => _eventContainer.ValidatingConnectionEvent.RemoveHandler(value); - } - - public event Func ClientConnectedAsync - { - add => _eventContainer.ClientConnectedEvent.AddHandler(value); - remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value); - } - - public event Func ClientDisconnectedAsync - { - add => _eventContainer.ClientDisconnectedEvent.AddHandler(value); - remove => _eventContainer.ClientDisconnectedEvent.RemoveHandler(value); - } - - public event Func InterceptingInboundPacketAsync - { - add => _eventContainer.InterceptingInboundPacketEvent.AddHandler(value); - remove => _eventContainer.InterceptingInboundPacketEvent.RemoveHandler(value); - } - - public event Func InterceptingOutboundPacketAsync - { - add => _eventContainer.InterceptingOutboundPacketEvent.AddHandler(value); - remove => _eventContainer.InterceptingOutboundPacketEvent.RemoveHandler(value); - } - - public event Func InterceptingSubscriptionAsync - { - add => _eventContainer.InterceptingSubscriptionEvent.AddHandler(value); - remove => _eventContainer.InterceptingSubscriptionEvent.RemoveHandler(value); - } - - public event Func InterceptingUnsubscriptionAsync - { - add => _eventContainer.InterceptingUnsubscriptionEvent.AddHandler(value); - remove => _eventContainer.InterceptingUnsubscriptionEvent.RemoveHandler(value); - } - - public event Func InterceptingPublishAsync - { - add => _eventContainer.InterceptingPublishEvent.AddHandler(value); - remove => _eventContainer.InterceptingPublishEvent.RemoveHandler(value); - } - - public event Func ClientSubscribedTopicAsync - { - add => _eventContainer.ClientSubscribedTopicEvent.AddHandler(value); - remove => _eventContainer.ClientSubscribedTopicEvent.RemoveHandler(value); - } - - public event Func ClientUnsubscribedTopicAsync - { - add => _eventContainer.ClientUnsubscribedTopicEvent.AddHandler(value); - remove => _eventContainer.ClientUnsubscribedTopicEvent.RemoveHandler(value); - } - - public event Func PreparingSessionAsync - { - add => _eventContainer.PreparingSessionEvent.AddHandler(value); - remove => _eventContainer.PreparingSessionEvent.RemoveHandler(value); } public event Func ApplicationMessageNotConsumedAsync @@ -123,27 +46,123 @@ namespace MQTTnet.Server remove => _eventContainer.ApplicationMessageNotConsumedEvent.RemoveHandler(value); } - public event Func RetainedMessageChangedAsync + public event Func ClientConnectedAsync { - add => _eventContainer.RetainedMessageChangedEvent.AddHandler(value); - remove => _eventContainer.RetainedMessageChangedEvent.RemoveHandler(value); + add => _eventContainer.ClientConnectedEvent.AddHandler(value); + remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value); } - + + public event Func ClientDisconnectedAsync + { + add => _eventContainer.ClientDisconnectedEvent.AddHandler(value); + remove => _eventContainer.ClientDisconnectedEvent.RemoveHandler(value); + } + + public event Func ClientSubscribedTopicAsync + { + add => _eventContainer.ClientSubscribedTopicEvent.AddHandler(value); + remove => _eventContainer.ClientSubscribedTopicEvent.RemoveHandler(value); + } + + public event Func ClientUnsubscribedTopicAsync + { + add => _eventContainer.ClientUnsubscribedTopicEvent.AddHandler(value); + remove => _eventContainer.ClientUnsubscribedTopicEvent.RemoveHandler(value); + } + + public event Func InterceptingInboundPacketAsync + { + add => _eventContainer.InterceptingInboundPacketEvent.AddHandler(value); + remove => _eventContainer.InterceptingInboundPacketEvent.RemoveHandler(value); + } + + public event Func InterceptingOutboundPacketAsync + { + add => _eventContainer.InterceptingOutboundPacketEvent.AddHandler(value); + remove => _eventContainer.InterceptingOutboundPacketEvent.RemoveHandler(value); + } + + public event Func InterceptingPublishAsync + { + add => _eventContainer.InterceptingPublishEvent.AddHandler(value); + remove => _eventContainer.InterceptingPublishEvent.RemoveHandler(value); + } + + public event Func InterceptingSubscriptionAsync + { + add => _eventContainer.InterceptingSubscriptionEvent.AddHandler(value); + remove => _eventContainer.InterceptingSubscriptionEvent.RemoveHandler(value); + } + + public event Func InterceptingUnsubscriptionAsync + { + add => _eventContainer.InterceptingUnsubscriptionEvent.AddHandler(value); + remove => _eventContainer.InterceptingUnsubscriptionEvent.RemoveHandler(value); + } + public event Func LoadingRetainedMessageAsync { add => _eventContainer.LoadingRetainedMessagesEvent.AddHandler(value); remove => _eventContainer.LoadingRetainedMessagesEvent.RemoveHandler(value); } - + + public event Func PreparingSessionAsync + { + add => _eventContainer.PreparingSessionEvent.AddHandler(value); + remove => _eventContainer.PreparingSessionEvent.RemoveHandler(value); + } + + public event Func RetainedMessageChangedAsync + { + add => _eventContainer.RetainedMessageChangedEvent.AddHandler(value); + remove => _eventContainer.RetainedMessageChangedEvent.RemoveHandler(value); + } + public event Func RetainedMessagesClearedAsync { add => _eventContainer.RetainedMessagesClearedEvent.AddHandler(value); remove => _eventContainer.RetainedMessagesClearedEvent.RemoveHandler(value); } + public event Func SessionDeletedAsync + { + add => _eventContainer.SessionDeletedEvent.AddHandler(value); + remove => _eventContainer.SessionDeletedEvent.RemoveHandler(value); + } + + public event Func StartedAsync + { + add => _eventContainer.StartedEvent.AddHandler(value); + remove => _eventContainer.StartedEvent.RemoveHandler(value); + } + + public event Func StoppedAsync + { + add => _eventContainer.StoppedEvent.AddHandler(value); + remove => _eventContainer.StoppedEvent.RemoveHandler(value); + } + + public event Func ValidatingConnectionAsync + { + add => _eventContainer.ValidatingConnectionEvent.AddHandler(value); + remove => _eventContainer.ValidatingConnectionEvent.RemoveHandler(value); + } + public bool IsStarted => _cancellationTokenSource != null; - - public Task> GetClientsAsync() + + /// + /// Deletes all retained messages. + /// + /// A task. + public Task DeleteRetainedMessagesAsync() + { + ThrowIfDisposed(); + ThrowIfNotStarted(); + + return _retainedMessagesManager?.ClearMessages() ?? PlatformAbstractionLayer.CompletedTask; + } + + public Task> GetClientsAsync() { ThrowIfDisposed(); ThrowIfNotStarted(); @@ -151,16 +170,8 @@ namespace MQTTnet.Server return _clientSessionsManager.GetClientStatusAsync(); } - public Task> GetSessionsAsync() - { - ThrowIfDisposed(); - ThrowIfNotStarted(); - - return _clientSessionsManager.GetSessionStatusAsync(); - } - /// - /// Gets all currently cached retained messages. + /// Gets all currently cached retained messages. /// /// A list of retained messages. public Task> GetRetainedMessagesAsync() @@ -172,80 +183,63 @@ namespace MQTTnet.Server } /// - /// Deletes all retained messages. + /// Disconnects the client with the specified ID. /// + /// The ID. + /// The reason code. /// A task. - public Task DeleteRetainedMessagesAsync() + public Task DisconnectClientAsync(string id, MqttDisconnectReasonCode reasonCode) { - ThrowIfDisposed(); - ThrowIfNotStarted(); - - return _retainedMessagesManager?.ClearMessages() ?? PlatformAbstractionLayer.CompletedTask; - } - - /// - /// Updates the specified retained message. - /// Setting the payload to null or empty will delete the retained message. - /// - /// The retained message. - /// A task. - public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage) - { - if (retainedMessage == null) throw new ArgumentNullException(nameof(retainedMessage)); - - ThrowIfDisposed(); - ThrowIfNotStarted(); - - return _retainedMessagesManager?.UpdateMessage(null, retainedMessage); - } - - public Task SubscribeAsync(string clientId, ICollection topicFilters) - { - if (clientId == null) throw new ArgumentNullException(nameof(clientId)); - if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - - foreach (var topicFilter in topicFilters) + // TODO: Add more options. + if (id == null) { - MqttTopicValidator.ThrowIfInvalidSubscribe(topicFilter.Topic); + throw new ArgumentNullException(nameof(id)); + } + + return _clientSessionsManager.GetClient(id).StopAsync(reasonCode); + } + + public Task> GetSessionsAsync() + { + ThrowIfDisposed(); + ThrowIfNotStarted(); + + return _clientSessionsManager.GetSessionStatusAsync(); + } + + public sealed class MqttInjectedApplicationMessage + { + public string SenderClientId { get; set; } + + public MqttApplicationMessage ApplicationMessage { get; set; } + } + + public Task InjectApplicationMessage(MqttInjectedApplicationMessage injectedApplicationMessage, CancellationToken cancellationToken = default) + { + if (injectedApplicationMessage == null) + { + throw new ArgumentNullException(nameof(injectedApplicationMessage)); + } + + if (injectedApplicationMessage.ApplicationMessage == null) + { + throw new ArgumentNullException(nameof(injectedApplicationMessage.ApplicationMessage)); } ThrowIfDisposed(); - ThrowIfNotStarted(); - return _clientSessionsManager.SubscribeAsync(clientId, topicFilters); - } - - public Task UnsubscribeAsync(string clientId, ICollection topicFilters) - { - if (clientId == null) throw new ArgumentNullException(nameof(clientId)); - if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - - ThrowIfDisposed(); - ThrowIfNotStarted(); - - return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters); - } - - public async Task PublishAsync(string senderClientId, MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default) - { - if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - - ThrowIfDisposed(); - - MqttTopicValidator.ThrowIfInvalid(applicationMessage.Topic); + MqttTopicValidator.ThrowIfInvalid(injectedApplicationMessage.ApplicationMessage.Topic); ThrowIfNotStarted(); - await _clientSessionsManager.DispatchPublishPacket(senderClientId, applicationMessage); - - return new MqttClientPublishResult(); + return _clientSessionsManager.DispatchPublishPacket(injectedApplicationMessage.SenderClientId, injectedApplicationMessage.ApplicationMessage); } - + public async Task StartAsync() { ThrowIfDisposed(); ThrowIfStarted(); - + _cancellationTokenSource = new CancellationTokenSource(); var cancellationToken = _cancellationTokenSource.Token; @@ -253,7 +247,7 @@ namespace MQTTnet.Server await _retainedMessagesManager.LoadMessages().ConfigureAwait(false); _clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, _eventContainer, _rootLogger); - + _keepAliveMonitor = new MqttServerKeepAliveMonitor(_options, _clientSessionsManager, _rootLogger); _keepAliveMonitor.Start(cancellationToken); @@ -280,7 +274,7 @@ namespace MQTTnet.Server _cancellationTokenSource.Cancel(false); await _clientSessionsManager.CloseAllConnectionsAsync().ConfigureAwait(false); - + foreach (var adapter in _adapters) { adapter.ClientHandler = null; @@ -303,6 +297,66 @@ namespace MQTTnet.Server await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); } + public Task SubscribeAsync(string clientId, ICollection topicFilters) + { + if (clientId == null) + { + throw new ArgumentNullException(nameof(clientId)); + } + + if (topicFilters == null) + { + throw new ArgumentNullException(nameof(topicFilters)); + } + + foreach (var topicFilter in topicFilters) + { + MqttTopicValidator.ThrowIfInvalidSubscribe(topicFilter.Topic); + } + + ThrowIfDisposed(); + ThrowIfNotStarted(); + + return _clientSessionsManager.SubscribeAsync(clientId, topicFilters); + } + + public Task UnsubscribeAsync(string clientId, ICollection topicFilters) + { + if (clientId == null) + { + throw new ArgumentNullException(nameof(clientId)); + } + + if (topicFilters == null) + { + throw new ArgumentNullException(nameof(topicFilters)); + } + + ThrowIfDisposed(); + ThrowIfNotStarted(); + + return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters); + } + + /// + /// Updates the specified retained message. + /// Setting the payload to null or empty will delete the retained message. + /// + /// The retained message. + /// A task. + public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage) + { + if (retainedMessage == null) + { + throw new ArgumentNullException(nameof(retainedMessage)); + } + + ThrowIfDisposed(); + ThrowIfNotStarted(); + + return _retainedMessagesManager?.UpdateMessage(null, retainedMessage); + } + protected override void Dispose(bool disposing) { if (disposing) @@ -323,14 +377,6 @@ namespace MQTTnet.Server return _clientSessionsManager.HandleClientConnectionAsync(channelAdapter, cancellationToken); } - void ThrowIfStarted() - { - if (_cancellationTokenSource != null) - { - throw new InvalidOperationException("The MQTT server is already started."); - } - } - void ThrowIfNotStarted() { if (_cancellationTokenSource == null) @@ -338,5 +384,13 @@ namespace MQTTnet.Server throw new InvalidOperationException("The MQTT server is not started."); } } + + void ThrowIfStarted() + { + if (_cancellationTokenSource != null) + { + throw new InvalidOperationException("The MQTT server is already started."); + } + } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttPendingMessagesOverflowStrategy.cs b/Source/MQTTnet/Server/Options/MqttPendingMessagesOverflowStrategy.cs similarity index 100% rename from Source/MQTTnet/Server/MqttPendingMessagesOverflowStrategy.cs rename to Source/MQTTnet/Server/Options/MqttPendingMessagesOverflowStrategy.cs diff --git a/Source/MQTTnet/Server/PublishResponse.cs b/Source/MQTTnet/Server/PublishResponse.cs index bc94e34a..0eae1aab 100644 --- a/Source/MQTTnet/Server/PublishResponse.cs +++ b/Source/MQTTnet/Server/PublishResponse.cs @@ -1,11 +1,12 @@ using System.Collections.Generic; using MQTTnet.Packets; +using MQTTnet.Protocol; namespace MQTTnet.Server { public sealed class PublishResponse { - public MqttApplicationMessageResponseReasonCode ReasonCode { get; set; } = MqttApplicationMessageResponseReasonCode.Success; + public MqttPubAckReasonCode ReasonCode { get; set; } = MqttPubAckReasonCode.Success; public string ReasonString { get; set; } diff --git a/Source/MQTTnet/Server/Status/IMqttClientStatus.cs b/Source/MQTTnet/Server/Status/IMqttClientStatus.cs deleted file mode 100644 index c48ca1ad..00000000 --- a/Source/MQTTnet/Server/Status/IMqttClientStatus.cs +++ /dev/null @@ -1,45 +0,0 @@ -using System; -using System.Threading.Tasks; -using MQTTnet.Formatter; - -namespace MQTTnet.Server -{ - public interface IMqttClientStatus - { - /// - /// Gets the client identifier. - /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. - /// - string ClientId { get; } - - string Endpoint { get; } - - MqttProtocolVersion ProtocolVersion { get; } - - DateTime ConnectedTimestamp { get; } - - DateTime LastPacketReceivedTimestamp { get; } - - DateTime LastPacketSentTimestamp { get; } - - DateTime LastNonKeepAlivePacketReceivedTimestamp { get; } - - long ReceivedApplicationMessagesCount { get; } - - long SentApplicationMessagesCount { get; } - - long ReceivedPacketsCount { get; } - - long SentPacketsCount { get; } - - IMqttSessionStatus Session { get; } - - long BytesSent { get; } - - long BytesReceived { get; } - - Task DisconnectAsync(); - - void ResetStatistics(); - } -} diff --git a/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs b/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs deleted file mode 100644 index ad39de1b..00000000 --- a/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs +++ /dev/null @@ -1,55 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Threading.Tasks; - -namespace MQTTnet.Server -{ - public interface IMqttSessionStatus - { - /// - /// Fired when this session is being deleted. - /// - event EventHandler Deleted; - - /// - /// Gets or sets the client identifier. - /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. - /// - string ClientId { get; } - - /// - /// Gets the count of messages which are not yet sent to the client but already queued. - /// - long PendingApplicationMessagesCount { get; } - - /// - /// This items can be used by the library user in order to store custom information. - /// - IDictionary Items { get; } - - /// - /// Enqueues a new application message. - /// - /// The application message. - /// A task. - Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage); - - /// - /// Delivers a new application message. - /// - /// The application message. - /// A task. - Task DeliverApplicationMessageAsync(MqttApplicationMessage applicationMessage); - - /// - /// Clears the entire queue with pending application messages. - /// - /// A task. - Task ClearApplicationMessagesQueueAsync(); - - [Obsolete("Please use _ClearApplicationMessagesQueueAsync_ instead. This will be removed soon.")] - Task ClearPendingApplicationMessagesAsync(); - - Task DeleteAsync(); - } -} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Status/MqttClientStatus.cs b/Source/MQTTnet/Server/Status/MqttClientStatus.cs index 03c5a487..f6f4560b 100644 --- a/Source/MQTTnet/Server/Status/MqttClientStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttClientStatus.cs @@ -1,59 +1,59 @@ using System; using System.Threading.Tasks; -using MQTTnet.Client; using MQTTnet.Formatter; +using MQTTnet.Protocol; namespace MQTTnet.Server { - public sealed class MqttClientStatus : IMqttClientStatus + public sealed class MqttClientStatus { - readonly MqttClient _connection; + readonly MqttClient _client; - public MqttClientStatus(MqttClient connection) + public MqttClientStatus(MqttClient client) { - _connection = connection ?? throw new ArgumentNullException(nameof(connection)); + _client = client ?? throw new ArgumentNullException(nameof(client)); } /// /// Gets or sets the client identifier. /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// - public string ClientId => _connection.Id; + public string Id => _client.Id; - public string Endpoint => _connection.Endpoint; + public string Endpoint => _client.Endpoint; - public MqttProtocolVersion ProtocolVersion => _connection.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion; + public MqttProtocolVersion ProtocolVersion => _client.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion; - public DateTime ConnectedTimestamp => _connection.Statistics.ConnectedTimestamp; + public DateTime ConnectedTimestamp => _client.Statistics.ConnectedTimestamp; - public DateTime LastPacketReceivedTimestamp => _connection.Statistics.LastPacketReceivedTimestamp; + public DateTime LastPacketReceivedTimestamp => _client.Statistics.LastPacketReceivedTimestamp; - public DateTime LastPacketSentTimestamp => _connection.Statistics.LastPacketSentTimestamp; + public DateTime LastPacketSentTimestamp => _client.Statistics.LastPacketSentTimestamp; - public DateTime LastNonKeepAlivePacketReceivedTimestamp => _connection.Statistics.LastNonKeepAlivePacketReceivedTimestamp; + public DateTime LastNonKeepAlivePacketReceivedTimestamp => _client.Statistics.LastNonKeepAlivePacketReceivedTimestamp; - public long ReceivedApplicationMessagesCount => _connection.Statistics.ReceivedApplicationMessagesCount; + public long ReceivedApplicationMessagesCount => _client.Statistics.ReceivedApplicationMessagesCount; - public long SentApplicationMessagesCount => _connection.Statistics.SentApplicationMessagesCount; + public long SentApplicationMessagesCount => _client.Statistics.SentApplicationMessagesCount; - public long ReceivedPacketsCount => _connection.Statistics.ReceivedPacketsCount; + public long ReceivedPacketsCount => _client.Statistics.ReceivedPacketsCount; - public long SentPacketsCount => _connection.Statistics.SentPacketsCount; + public long SentPacketsCount => _client.Statistics.SentPacketsCount; - public IMqttSessionStatus Session { get; set; } + public MqttSessionStatus Session { get; set; } - public long BytesSent => _connection.ChannelAdapter.BytesSent; + public long BytesSent => _client.ChannelAdapter.BytesSent; - public long BytesReceived => _connection.ChannelAdapter.BytesReceived; + public long BytesReceived => _client.ChannelAdapter.BytesReceived; public Task DisconnectAsync() { - return _connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection); + return _client.StopAsync(MqttDisconnectReasonCode.NormalDisconnection); } public void ResetStatistics() { - _connection.ResetStatistics(); + _client.ResetStatistics(); } } } diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index c7a5d990..2c5e914a 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -1,5 +1,5 @@ using System; -using System.Collections.Generic; +using System.Collections; using System.Threading.Tasks; using MQTTnet.Formatter; using MQTTnet.Implementations; @@ -7,7 +7,7 @@ using MQTTnet.Internal; namespace MQTTnet.Server { - public sealed class MqttSessionStatus : IMqttSessionStatus + public sealed class MqttSessionStatus { readonly MqttSession _session; @@ -16,33 +16,18 @@ namespace MQTTnet.Server _session = session ?? throw new ArgumentNullException(nameof(session)); } - public event EventHandler Deleted - { - add => _session.Deleted += value; - remove => _session.Deleted += value; - } - - public string ClientId => _session.Id; + public string Id => _session.Id; public long PendingApplicationMessagesCount => _session.PendingDataPacketsCount; public DateTime CreatedTimestamp => _session.CreatedTimestamp; - public IDictionary Items => _session.Items; + public IDictionary Items => _session.Items; public Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage) { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - // _session.ApplicationMessagesQueue.Enqueue(new MqttQueuedApplicationMessage - // { - // ApplicationMessage = applicationMessage, - // IsDuplicate = false, - // IsRetainedMessage = false, - // SubscriptionQualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, - // SenderClientId = null - // }); - var publishPacketFactory = new MqttPublishPacketFactory(); _session.EnqueuePacket(new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage))); @@ -59,22 +44,15 @@ namespace MQTTnet.Server return packetBusItem.WaitForDeliveryAsync(); } - - public Task ClearApplicationMessagesQueueAsync() - { - // TODO: Fix! - //_session.ApplicationMessagesQueue.Clear(); - return Task.FromResult(0); - } - + public Task DeleteAsync() { return _session.DeleteAsync(); } - public Task ClearPendingApplicationMessagesAsync() + public Task ClearApplicationMessagesQueueAsync() { - return ClearApplicationMessagesQueueAsync(); + throw new NotImplementedException(); } } } diff --git a/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs index 2e8dc4f7..bf4cf3ad 100644 --- a/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs @@ -2,6 +2,7 @@ using BenchmarkDotNet.Jobs; using MQTTnet.Client; using MQTTnet.Server; +using IMqttClient = MQTTnet.Client.IMqttClient; namespace MQTTnet.Benchmarks { diff --git a/Tests/MQTTnet.Core.Tests/Client/ManagedMqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/Client/ManagedMqttClient_Tests.cs index b653a4bc..5164317f 100644 --- a/Tests/MQTTnet.Core.Tests/Client/ManagedMqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Client/ManagedMqttClient_Tests.cs @@ -9,6 +9,7 @@ using MQTTnet.Diagnostics; using MQTTnet.Extensions.ManagedClient; using MQTTnet.Server; using MQTTnet.Tests.Mockups; +using IMqttClient = MQTTnet.Client.IMqttClient; namespace MQTTnet.Tests.Client { diff --git a/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs b/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs index aa177f91..b90e84ab 100644 --- a/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs @@ -18,7 +18,7 @@ namespace MQTTnet.Tests.Internal using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1))) { - await bus.DequeueAsync(timeout.Token); + await bus.DequeueItemAsync(timeout.Token); } } @@ -27,31 +27,31 @@ namespace MQTTnet.Tests.Internal { var bus = new MqttPacketBus(); - bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - bus.Enqueue(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); - bus.Enqueue(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); - bus.Enqueue(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); + bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); + bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); + bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control); - bus.Enqueue(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); - bus.Enqueue(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); - bus.Enqueue(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); - Assert.AreEqual(9, bus.PacketsCount); + Assert.AreEqual(9, bus.ItemsCount); - Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); - Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); - Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); - Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); - Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); - Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); - Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); - Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); - Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); + Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); + Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); + Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); + Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); + Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); + Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); + Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); + Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); + Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); - Assert.AreEqual(0, bus.PacketsCount); + Assert.AreEqual(0, bus.ItemsCount); } [TestMethod] @@ -59,11 +59,11 @@ namespace MQTTnet.Tests.Internal { var bus = new MqttPacketBus(); - bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); + bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - Assert.AreEqual(3, bus.PacketsCount); + Assert.AreEqual(3, bus.ItemsCount); var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control); Assert.AreEqual(0, exportedPackets.Count); @@ -74,7 +74,7 @@ namespace MQTTnet.Tests.Internal exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data); Assert.AreEqual(3, exportedPackets.Count); - Assert.AreEqual(3, bus.PacketsCount); + Assert.AreEqual(3, bus.ItemsCount); } [TestMethod] @@ -93,21 +93,21 @@ namespace MQTTnet.Tests.Internal delivered = true; }; - bus.Enqueue(item1, MqttPacketBusPartition.Data); - bus.Enqueue(item2, MqttPacketBusPartition.Data); - bus.Enqueue(item3, MqttPacketBusPartition.Data); + bus.EnqueueItem(item1, MqttPacketBusPartition.Data); + bus.EnqueueItem(item2, MqttPacketBusPartition.Data); + bus.EnqueueItem(item3, MqttPacketBusPartition.Data); Assert.IsFalse(delivered); - bus.DequeueAsync(CancellationToken.None).Result.MarkAsDelivered(); + bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered(); Assert.IsFalse(delivered); - bus.DequeueAsync(CancellationToken.None).Result.MarkAsDelivered(); + bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered(); Assert.IsFalse(delivered); - bus.DequeueAsync(CancellationToken.None).Result.MarkAsDelivered(); + bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered(); // The third packet has the event attached. Assert.IsTrue(delivered); diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index c0a93548..247bbc0d 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs @@ -12,6 +12,7 @@ using MQTTnet.Extensions.Rpc; using MQTTnet.Extensions.Rpc.Options; using MQTTnet.Formatter; using MQTTnet.LowLevelClient; +using IMqttClient = MQTTnet.Client.IMqttClient; namespace MQTTnet.Tests.Mockups { diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index 05c8e3bb..a930c62e 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -11,390 +11,144 @@ using MQTTnet.Formatter; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server; -using MQTTnet.Tests.Mockups; namespace MQTTnet.Tests.Server { [TestClass] public sealed class General_Tests : BaseTestClass { + Dictionary _connected; + [TestMethod] - [DataRow("", null)] - [DataRow("", "")] - [DataRow(null, null)] - public async Task Use_Admissible_Credentials(string username, string password) + public async Task Client_Disconnect_Without_Errors() { using (var testEnvironment = CreateTestEnvironment()) { - await testEnvironment.StartServer(); + bool clientWasConnected; - var client = testEnvironment.CreateClient(); - - var clientOptions = new MqttClientOptionsBuilder() - .WithTcpServer("localhost", testEnvironment.ServerPort) - .WithCredentials(username, password) - .Build(); - - var connectResult = await client.ConnectAsync(clientOptions); - - Assert.IsFalse(connectResult.IsSessionPresent); - Assert.IsTrue(client.IsConnected); - } - } - - [TestMethod] - public async Task Use_Empty_Client_ID() - { - using (var testEnvironment = CreateTestEnvironment()) - { - await testEnvironment.StartServer(); - - var client = testEnvironment.CreateClient(); - - var clientOptions = new MqttClientOptionsBuilder() - .WithTcpServer("localhost", testEnvironment.ServerPort) - .WithClientId(string.Empty) - .Build(); - - var connectResult = await client.ConnectAsync(clientOptions); - - Assert.IsFalse(connectResult.IsSessionPresent); - Assert.IsTrue(client.IsConnected); - } - } - - [TestMethod] - public async Task Publish_At_Most_Once_0x00() - { - await TestPublishAsync( - "A/B/C", - MqttQualityOfServiceLevel.AtMostOnce, - "A/B/C", - MqttQualityOfServiceLevel.AtMostOnce, - 1); - } - - [TestMethod] - public async Task Publish_At_Least_Once_0x01() - { - await TestPublishAsync( - "A/B/C", - MqttQualityOfServiceLevel.AtLeastOnce, - "A/B/C", - MqttQualityOfServiceLevel.AtLeastOnce, - 1); - } - - [TestMethod] - public async Task Publish_Exactly_Once_0x02() - { - await TestPublishAsync( - "A/B/C", - MqttQualityOfServiceLevel.ExactlyOnce, - "A/B/C", - MqttQualityOfServiceLevel.ExactlyOnce, - 1); - } - - [TestMethod] - public async Task Use_Clean_Session() - { - using (var testEnvironment = CreateTestEnvironment()) - { - await testEnvironment.StartServer(); - - var client = testEnvironment.CreateClient(); - var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build()); - - Assert.IsFalse(connectResult.IsSessionPresent); - } - } - - [TestMethod] - public async Task Will_Message_Do_Not_Send_On_Clean_Disconnect() - { - using (var testEnvironment = CreateTestEnvironment()) - { - var receivedMessagesCount = 0; - - await testEnvironment.StartServer(); - - var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").Build(); - - var clientOptions = new MqttClientOptionsBuilder().WithWillMessage(willMessage); - - var c1 = await testEnvironment.ConnectClient(); - - c1.ApplicationMessageReceivedAsync += e => + var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder()); + try { - Interlocked.Increment(ref receivedMessagesCount); - return Task.CompletedTask; - }; - - await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); + var client = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder()); - var c2 = await testEnvironment.ConnectClient(clientOptions); - await c2.DisconnectAsync().ConfigureAwait(false); + clientWasConnected = true; - await Task.Delay(1000); + await client.DisconnectAsync(); - Assert.AreEqual(0, receivedMessagesCount); - } - } - - [TestMethod] - public async Task Will_Message_Do_Not_Send_On_Takeover() - { - using (var testEnvironment = CreateTestEnvironment()) - { - var receivedMessagesCount = 0; - - await testEnvironment.StartServer(); - - // C1 will receive the last will! - var c1 = await testEnvironment.ConnectClient(); - c1.ApplicationMessageReceivedAsync += e => + await Task.Delay(500); + } + finally { - Interlocked.Increment(ref receivedMessagesCount); - return Task.CompletedTask; - }; - - await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); - - // C2 has the last will defined. - var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").Build(); - - var clientOptions = new MqttClientOptionsBuilder() - .WithWillMessage(willMessage) - .WithClientId("WillOwner"); - - var c2 = await testEnvironment.ConnectClient(clientOptions); - - // C3 will do the connection takeover. - var c3 = await testEnvironment.ConnectClient(clientOptions); - - await Task.Delay(1000); - - Assert.AreEqual(0, receivedMessagesCount); - } - } - - [TestMethod] - public async Task Will_Message_Send() - { - using (var testEnvironment = CreateTestEnvironment()) - { - await testEnvironment.StartServer(); - - var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build(); - - var clientOptions = new MqttClientOptionsBuilder().WithWillMessage(willMessage); - - var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder()); - - var receivedMessagesCount = 0; - c1.ApplicationMessageReceivedAsync += e => - { - Interlocked.Increment(ref receivedMessagesCount); - return Task.CompletedTask; - }; - - await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); - - var c2 = await testEnvironment.ConnectClient(clientOptions); - c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent. - - await Task.Delay(1000); - - Assert.AreEqual(1, receivedMessagesCount); - } - } - - [TestMethod] - public async Task Publish_From_Server() - { - using (var testEnvironment = CreateTestEnvironment()) - { - var server = await testEnvironment.StartServer(); - - var receivedMessagesCount = 0; - - var client = await testEnvironment.ConnectClient(); - client.ApplicationMessageReceivedAsync += e => - { - Interlocked.Increment(ref receivedMessagesCount); - return Task.CompletedTask; - }; - - var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); - await client.SubscribeAsync(new MqttTopicFilter {Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce}); - - await server.PublishAsync("server", message); - - await Task.Delay(1000); - - Assert.AreEqual(1, receivedMessagesCount); - } - } - - [TestMethod] - public async Task Publish_Multiple_Clients() - { - var receivedMessagesCount = 0; - - using (var testEnvironment = CreateTestEnvironment()) - { - await testEnvironment.StartServer(); - - var c1 = await testEnvironment.ConnectClient(); - var c2 = await testEnvironment.ConnectClient(); - var c3 = await testEnvironment.ConnectClient(); - - c2.ApplicationMessageReceivedAsync += e => - { - Interlocked.Increment(ref receivedMessagesCount); - return Task.CompletedTask; - }; - - c3.ApplicationMessageReceivedAsync += e => - { - Interlocked.Increment(ref receivedMessagesCount); - return Task.CompletedTask; - }; - - await c2.SubscribeAsync(new MqttTopicFilter {Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce}).ConfigureAwait(false); - await c3.SubscribeAsync(new MqttTopicFilter {Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce}).ConfigureAwait(false); - - var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); - - for (var i = 0; i < 500; i++) - { - await c1.PublishAsync(message).ConfigureAwait(false); + await server.StopAsync(); } - SpinWait.SpinUntil(() => receivedMessagesCount == 1000, TimeSpan.FromSeconds(20)); + Assert.IsTrue(clientWasConnected); - Assert.AreEqual(1000, receivedMessagesCount); + testEnvironment.ThrowIfLogErrors(); } } [TestMethod] - public async Task Session_Takeover() + public async Task Collect_Messages_In_Disconnected_Session() { using (var testEnvironment = CreateTestEnvironment()) { - await testEnvironment.StartServer(); + var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithPersistentSessions()); - var options = new MqttClientOptionsBuilder() - .WithCleanSession(false) - .WithProtocolVersion(MqttProtocolVersion.V500) // Disconnect reason is only available in MQTT 5+ - .WithClientId("a"); - - var client1 = await testEnvironment.ConnectClient(options); + // Create the session including the subscription. + var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a").WithCleanSession(false)); + await client1.SubscribeAsync("x"); + await client1.DisconnectAsync(); await Task.Delay(500); - MqttClientDisconnectReason disconnectReason = MqttClientDisconnectReason.NormalDisconnection; - client1.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(c => { disconnectReason = c.Reason; }); + var clientStatus = await server.GetClientsAsync(); + Assert.AreEqual(0, clientStatus.Count); + + var client2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("b").WithCleanSession(false)); + await client2.PublishAsync("x", "1"); + await client2.PublishAsync("x", "2"); + await client2.PublishAsync("x", "3"); + await client2.DisconnectAsync(); - var client2 = await testEnvironment.ConnectClient(options); await Task.Delay(500); - Assert.IsFalse(client1.IsConnected); - Assert.IsTrue(client2.IsConnected); + clientStatus = await server.GetClientsAsync(); + var sessionStatus = await server.GetSessionsAsync(); - Assert.AreEqual(MqttClientDisconnectReason.SessionTakenOver, disconnectReason); + Assert.AreEqual(0, clientStatus.Count); + Assert.AreEqual(2, sessionStatus.Count); + + Assert.AreEqual(3, sessionStatus.First(s => s.Id == client1.Options.ClientId).PendingApplicationMessagesCount); } } [TestMethod] - public async Task No_Messages_If_No_Subscription() + public async Task Deny_Connection() { using (var testEnvironment = CreateTestEnvironment()) { - await testEnvironment.StartServer(); + testEnvironment.IgnoreClientLogErrors = true; - var client = await testEnvironment.ConnectClient(); - var receivedMessages = new List(); + var server = await testEnvironment.StartServer(); - client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(async e => { await client.PublishAsync("Connected"); }); - - client.ApplicationMessageReceivedAsync += e => + server.ValidatingConnectionAsync += e => { - lock (receivedMessages) - { - receivedMessages.Add(e.ApplicationMessage); - } - + e.ReasonCode = MqttConnectReasonCode.NotAuthorized; return Task.CompletedTask; }; - await Task.Delay(500); - - await client.PublishAsync("Hello"); - - await Task.Delay(500); - - Assert.AreEqual(0, receivedMessages.Count); + var connectingFailedException = await Assert.ThrowsExceptionAsync(() => testEnvironment.ConnectClient()); + Assert.AreEqual(MqttClientConnectResultCode.NotAuthorized, connectingFailedException.ResultCode); } } [TestMethod] - public async Task Set_Subscription_At_Server() + public async Task Do_Not_Send_Retained_Messages_For_Denied_Subscription() { using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(); - server.ClientConnectedAsync += async e => + server.InterceptingSubscriptionAsync += e => { - // Every client will automatically subscribe to this topic. - await server.SubscribeAsync(e.ClientId, "topic1"); - }; - - var client = await testEnvironment.ConnectClient(); - var receivedMessages = new List(); - - client.ApplicationMessageReceivedAsync += e => - { - lock (receivedMessages) + // This should lead to no subscriptions for "n" at all. So also no sending of retained messages. + if (e.TopicFilter.Topic == "n") { - receivedMessages.Add(e.ApplicationMessage); + e.Response.ReasonCode = MqttSubscribeReasonCode.UnspecifiedError; } return Task.CompletedTask; }; + // Prepare some retained messages. + var client1 = await testEnvironment.ConnectClient(); + await client1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("y").WithPayload("x").WithRetainFlag().Build()); + await client1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("n").WithPayload("x").WithRetainFlag().Build()); + await client1.DisconnectAsync(); + await Task.Delay(500); - await client.PublishAsync("Hello"); - await Task.Delay(100); - Assert.AreEqual(0, receivedMessages.Count); + // Subscribe to all retained message types. + // It is important to do this in a range of filters to ensure that a subscription is not "hidden". + var client2 = await testEnvironment.ConnectClient(); - await client.PublishAsync("topic1"); - await Task.Delay(100); - Assert.AreEqual(1, receivedMessages.Count); - } - } + var buffer = new StringBuilder(); - [TestMethod] - public async Task Shutdown_Disconnects_Clients_Gracefully() - { - using (var testEnvironment = CreateTestEnvironment()) - { - var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder()); + client2.ApplicationMessageReceivedAsync += e => + { + lock (buffer) + { + buffer.Append(e.ApplicationMessage.Topic); + } - var disconnectCalled = 0; + return Task.CompletedTask; + }; - var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder()); - c1.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(e => disconnectCalled++); + await client2.SubscribeAsync(new MqttTopicFilter {Topic = "y"}, new MqttTopicFilter {Topic = "n"}); - await Task.Delay(100); + await Task.Delay(500); - await server.StopAsync(); - - await Task.Delay(100); - - Assert.AreEqual(1, disconnectCalled); + Assert.AreEqual("y", buffer.ToString()); } } @@ -438,35 +192,6 @@ namespace MQTTnet.Tests.Server } } - [TestMethod] - public async Task Client_Disconnect_Without_Errors() - { - using (var testEnvironment = CreateTestEnvironment()) - { - bool clientWasConnected; - - var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder()); - try - { - var client = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder()); - - clientWasConnected = true; - - await client.DisconnectAsync(); - - await Task.Delay(500); - } - finally - { - await server.StopAsync(); - } - - Assert.IsTrue(clientWasConnected); - - testEnvironment.ThrowIfLogErrors(); - } - } - [TestMethod] public async Task Handle_Lots_Of_Parallel_Retained_Messages() { @@ -540,7 +265,7 @@ namespace MQTTnet.Tests.Server string receivedTopic = null; var c1 = await testEnvironment.ConnectClient(); await c1.SubscribeAsync("#"); - + c1.ApplicationMessageReceivedAsync += a => { receivedTopic = a.ApplicationMessage.Topic; @@ -554,68 +279,6 @@ namespace MQTTnet.Tests.Server } } - [TestMethod] - public async Task Persist_Retained_Message() - { - using (var testEnvironment = CreateTestEnvironment()) - { - List savedRetainedMessages = null; - - var s = await testEnvironment.StartServer(); - s.RetainedMessageChangedAsync += e => - { - savedRetainedMessages = e.StoredRetainedMessages; - return Task.CompletedTask; - }; - - var c1 = await testEnvironment.ConnectClient(); - - await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); - - await Task.Delay(500); - - Assert.AreEqual(1, savedRetainedMessages?.Count); - } - } - - [TestMethod] - public async Task Publish_After_Client_Connects() - { - using (var testEnvironment = CreateTestEnvironment()) - { - var server = await testEnvironment.StartServer(); - server.ClientConnectedAsync += async e => - { - await server.PublishAsync("server", new MqttApplicationMessage - { - Topic = "/test/1", - Payload = Encoding.UTF8.GetBytes("true"), - QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce - }); - }; - - string receivedTopic = null; - - var c1 = await testEnvironment.ConnectClient(); - c1.ApplicationMessageReceivedAsync += e => - { - receivedTopic = e.ApplicationMessage.Topic; - return Task.CompletedTask; - }; - - await c1.SubscribeAsync("#"); - - await testEnvironment.ConnectClient(); - await testEnvironment.ConnectClient(); - await testEnvironment.ConnectClient(); - await testEnvironment.ConnectClient(); - - await Task.Delay(500); - - Assert.AreEqual("/test/1", receivedTopic); - } - } - [TestMethod] public async Task Intercept_Message() { @@ -636,7 +299,7 @@ namespace MQTTnet.Tests.Server c2.ApplicationMessageReceivedAsync += e => { isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(e.ApplicationMessage.Payload), StringComparison.Ordinal) == 0; - + return Task.CompletedTask; }; @@ -650,82 +313,341 @@ namespace MQTTnet.Tests.Server } [TestMethod] - public async Task Send_Long_Body() + public async Task Intercept_Undelivered() { using (var testEnvironment = CreateTestEnvironment()) { - const int PayloadSizeInMB = 30; - const int CharCount = PayloadSizeInMB * 1024 * 1024; + var undeliverd = string.Empty; - var longBody = new byte[CharCount]; - byte @char = 32; - - for (long i = 0; i < PayloadSizeInMB * 1024L * 1024L; i++) + var server = await testEnvironment.StartServer(); + server.ApplicationMessageNotConsumedAsync += e => { - longBody[i] = @char; - - @char++; - - if (@char > 126) - { - @char = 32; - } - } - - byte[] receivedBody = null; - - await testEnvironment.StartServer(); - - var client1 = await testEnvironment.ConnectClient(); - client1.ApplicationMessageReceivedAsync += e => - { - receivedBody = e.ApplicationMessage.Payload; + undeliverd = e.ApplicationMessage.Topic; return Task.CompletedTask; }; - await client1.SubscribeAsync("string"); + var client = await testEnvironment.ConnectClient(); - var client2 = await testEnvironment.ConnectClient(); - await client2.PublishAsync("string", longBody); + await client.SubscribeAsync("b"); - await Task.Delay(TimeSpan.FromSeconds(5)); + await client.PublishAsync("a", null, MqttQualityOfServiceLevel.ExactlyOnce); - Assert.IsTrue(longBody.SequenceEqual(receivedBody ?? new byte[0])); + await Task.Delay(500); + + Assert.AreEqual("a", undeliverd); } } [TestMethod] - public async Task Deny_Connection() + public async Task No_Messages_If_No_Subscription() { using (var testEnvironment = CreateTestEnvironment()) { - testEnvironment.IgnoreClientLogErrors = true; + await testEnvironment.StartServer(); - var server = await testEnvironment.StartServer(); + var client = await testEnvironment.ConnectClient(); + var receivedMessages = new List(); - server.ValidatingConnectionAsync += e => + client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(async e => { await client.PublishAsync("Connected"); }); + + client.ApplicationMessageReceivedAsync += e => { - e.ReasonCode = MqttConnectReasonCode.NotAuthorized; + lock (receivedMessages) + { + receivedMessages.Add(e.ApplicationMessage); + } + return Task.CompletedTask; }; - var connectingFailedException = await Assert.ThrowsExceptionAsync(() => testEnvironment.ConnectClient()); - Assert.AreEqual(MqttClientConnectResultCode.NotAuthorized, connectingFailedException.ResultCode); + await Task.Delay(500); + + await client.PublishAsync("Hello"); + + await Task.Delay(500); + + Assert.AreEqual(0, receivedMessages.Count); } } - Dictionary _connected; - - void ConnectionValidationHandler(ValidatingConnectionEventArgs eventArgs) + [TestMethod] + public async Task Persist_Retained_Message() { - if (_connected.ContainsKey(eventArgs.ClientId)) + using (var testEnvironment = CreateTestEnvironment()) { - eventArgs.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; - return; - } + List savedRetainedMessages = null; - _connected[eventArgs.ClientId] = true; - eventArgs.ReasonCode = MqttConnectReasonCode.Success; + var s = await testEnvironment.StartServer(); + s.RetainedMessageChangedAsync += e => + { + savedRetainedMessages = e.StoredRetainedMessages; + return Task.CompletedTask; + }; + + var c1 = await testEnvironment.ConnectClient(); + + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); + + await Task.Delay(500); + + Assert.AreEqual(1, savedRetainedMessages?.Count); + } + } + + [TestMethod] + public async Task Publish_After_Client_Connects() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(); + server.ClientConnectedAsync += async e => + { + await server.InjectApplicationMessage(new MqttServer.MqttInjectedApplicationMessage + { + SenderClientId = "server", ApplicationMessage = new MqttApplicationMessage + { + Topic = "/test/1", + Payload = Encoding.UTF8.GetBytes("true"), + QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce + } + }); + }; + + string receivedTopic = null; + + var c1 = await testEnvironment.ConnectClient(); + c1.ApplicationMessageReceivedAsync += e => + { + receivedTopic = e.ApplicationMessage.Topic; + return Task.CompletedTask; + }; + + await c1.SubscribeAsync("#"); + + await testEnvironment.ConnectClient(); + await testEnvironment.ConnectClient(); + await testEnvironment.ConnectClient(); + await testEnvironment.ConnectClient(); + + await Task.Delay(500); + + Assert.AreEqual("/test/1", receivedTopic); + } + } + + [TestMethod] + public async Task Publish_At_Least_Once_0x01() + { + await TestPublishAsync( + "A/B/C", + MqttQualityOfServiceLevel.AtLeastOnce, + "A/B/C", + MqttQualityOfServiceLevel.AtLeastOnce, + 1); + } + + [TestMethod] + public async Task Publish_At_Most_Once_0x00() + { + await TestPublishAsync( + "A/B/C", + MqttQualityOfServiceLevel.AtMostOnce, + "A/B/C", + MqttQualityOfServiceLevel.AtMostOnce, + 1); + } + + [TestMethod] + public async Task Publish_Exactly_Once_0x02() + { + await TestPublishAsync( + "A/B/C", + MqttQualityOfServiceLevel.ExactlyOnce, + "A/B/C", + MqttQualityOfServiceLevel.ExactlyOnce, + 1); + } + + [TestMethod] + public async Task Publish_From_Server() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(); + + var receivedMessagesCount = 0; + + var client = await testEnvironment.ConnectClient(); + client.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return Task.CompletedTask; + }; + + var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); + await client.SubscribeAsync(new MqttTopicFilter {Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce}); + + await server.InjectApplicationMessage(new MqttServer.MqttInjectedApplicationMessage + { + SenderClientId = "server", + ApplicationMessage = message + }); + + await Task.Delay(1000); + + Assert.AreEqual(1, receivedMessagesCount); + } + } + + [TestMethod] + public async Task Publish_Multiple_Clients() + { + var receivedMessagesCount = 0; + + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var c1 = await testEnvironment.ConnectClient(); + var c2 = await testEnvironment.ConnectClient(); + var c3 = await testEnvironment.ConnectClient(); + + c2.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return Task.CompletedTask; + }; + + c3.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return Task.CompletedTask; + }; + + await c2.SubscribeAsync(new MqttTopicFilter {Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce}).ConfigureAwait(false); + await c3.SubscribeAsync(new MqttTopicFilter {Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce}).ConfigureAwait(false); + + var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); + + for (var i = 0; i < 500; i++) + { + await c1.PublishAsync(message).ConfigureAwait(false); + } + + SpinWait.SpinUntil(() => receivedMessagesCount == 1000, TimeSpan.FromSeconds(20)); + + Assert.AreEqual(1000, receivedMessagesCount); + } + } + + [TestMethod] + public async Task Remove_Session() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder()); + + var clientOptions = new MqttClientOptionsBuilder(); + var c1 = await testEnvironment.ConnectClient(clientOptions); + await Task.Delay(500); + Assert.AreEqual(1, (await server.GetClientsAsync()).Count); + + await c1.DisconnectAsync(); + await Task.Delay(500); + + Assert.AreEqual(0, (await server.GetClientsAsync()).Count); + } + } + + [TestMethod] + public async Task Same_Client_Id_Connect_Disconnect_Event_Order() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(); + + var events = new List(); + + server.ClientConnectedAsync += e => + { + lock (events) + { + events.Add("c"); + } + + return Task.CompletedTask; + }; + + server.ClientDisconnectedAsync += e => + { + lock (events) + { + events.Add("d"); + } + + return Task.CompletedTask; + }; + + var clientOptionsBuilder = new MqttClientOptionsBuilder() + .WithClientId(Guid.NewGuid().ToString()); + + // c + var c1 = await testEnvironment.ConnectClient(clientOptionsBuilder); + + await Task.Delay(500); + + var flow = string.Join(string.Empty, events); + Assert.AreEqual("c", flow); + + // dc + // Connect client with same client ID. Should disconnect existing client. + var c2 = await testEnvironment.ConnectClient(clientOptionsBuilder); + + await Task.Delay(500); + + flow = string.Join(string.Empty, events); + + Assert.AreEqual("cdc", flow); + + c2.ApplicationMessageReceivedAsync += e => + { + lock (events) + { + events.Add("r"); + } + + return Task.CompletedTask; + }; + + await c2.SubscribeAsync("topic"); + + // r + await c2.PublishAsync("topic"); + + await Task.Delay(500); + + flow = string.Join(string.Empty, events); + Assert.AreEqual("cdcr", flow); + + // nothing + + Assert.AreEqual(false, c1.IsConnected); + await c1.DisconnectAsync(); + Assert.AreEqual(false, c1.IsConnected); + + await Task.Delay(500); + + // d + Assert.AreEqual(true, c2.IsConnected); + await c2.DisconnectAsync(); + + await Task.Delay(500); + + await server.StopAsync(); + + flow = string.Join(string.Empty, events); + Assert.AreEqual("cdcrd", flow); + } } [TestMethod] @@ -744,7 +666,7 @@ namespace MQTTnet.Tests.Server ConnectionValidationHandler(e); return Task.CompletedTask; }; - + var events = new List(); server.ClientConnectedAsync += e => @@ -828,112 +750,135 @@ namespace MQTTnet.Tests.Server } [TestMethod] - public async Task Same_Client_Id_Connect_Disconnect_Event_Order() + public async Task Send_Long_Body() + { + using (var testEnvironment = CreateTestEnvironment()) + { + const int PayloadSizeInMB = 30; + const int CharCount = PayloadSizeInMB * 1024 * 1024; + + var longBody = new byte[CharCount]; + byte @char = 32; + + for (long i = 0; i < PayloadSizeInMB * 1024L * 1024L; i++) + { + longBody[i] = @char; + + @char++; + + if (@char > 126) + { + @char = 32; + } + } + + byte[] receivedBody = null; + + await testEnvironment.StartServer(); + + var client1 = await testEnvironment.ConnectClient(); + client1.ApplicationMessageReceivedAsync += e => + { + receivedBody = e.ApplicationMessage.Payload; + return Task.CompletedTask; + }; + + await client1.SubscribeAsync("string"); + + var client2 = await testEnvironment.ConnectClient(); + await client2.PublishAsync("string", longBody); + + await Task.Delay(TimeSpan.FromSeconds(5)); + + Assert.IsTrue(longBody.SequenceEqual(receivedBody ?? new byte[0])); + } + } + + [TestMethod] + public async Task Session_Takeover() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var options = new MqttClientOptionsBuilder() + .WithCleanSession(false) + .WithProtocolVersion(MqttProtocolVersion.V500) // Disconnect reason is only available in MQTT 5+ + .WithClientId("a"); + + var client1 = await testEnvironment.ConnectClient(options); + await Task.Delay(500); + + var disconnectReason = MqttClientDisconnectReason.NormalDisconnection; + client1.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(c => { disconnectReason = c.Reason; }); + + var client2 = await testEnvironment.ConnectClient(options); + await Task.Delay(500); + + Assert.IsFalse(client1.IsConnected); + Assert.IsTrue(client2.IsConnected); + + Assert.AreEqual(MqttClientDisconnectReason.SessionTakenOver, disconnectReason); + } + } + + [TestMethod] + public async Task Set_Subscription_At_Server() { using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(); - var events = new List(); - - server.ClientConnectedAsync += e => + server.ClientConnectedAsync += async e => { - lock (events) + // Every client will automatically subscribe to this topic. + await server.SubscribeAsync(e.ClientId, "topic1"); + }; + + var client = await testEnvironment.ConnectClient(); + var receivedMessages = new List(); + + client.ApplicationMessageReceivedAsync += e => + { + lock (receivedMessages) { - events.Add("c"); + receivedMessages.Add(e.ApplicationMessage); } return Task.CompletedTask; }; - server.ClientDisconnectedAsync += e => - { - lock (events) - { - events.Add("d"); - } - - return Task.CompletedTask; - }; - - var clientOptionsBuilder = new MqttClientOptionsBuilder() - .WithClientId(Guid.NewGuid().ToString()); - - // c - var c1 = await testEnvironment.ConnectClient(clientOptionsBuilder); - await Task.Delay(500); - var flow = string.Join(string.Empty, events); - Assert.AreEqual("c", flow); + await client.PublishAsync("Hello"); + await Task.Delay(100); + Assert.AreEqual(0, receivedMessages.Count); - // dc - // Connect client with same client ID. Should disconnect existing client. - var c2 = await testEnvironment.ConnectClient(clientOptionsBuilder); - - await Task.Delay(500); - - flow = string.Join(string.Empty, events); - - Assert.AreEqual("cdc", flow); - - c2.ApplicationMessageReceivedAsync += e => - { - lock (events) - { - events.Add("r"); - } - - return Task.CompletedTask; - }; - - await c2.SubscribeAsync("topic"); - - // r - await c2.PublishAsync("topic"); - - await Task.Delay(500); - - flow = string.Join(string.Empty, events); - Assert.AreEqual("cdcr", flow); - - // nothing - - Assert.AreEqual(false, c1.IsConnected); - await c1.DisconnectAsync(); - Assert.AreEqual(false, c1.IsConnected); - - await Task.Delay(500); - - // d - Assert.AreEqual(true, c2.IsConnected); - await c2.DisconnectAsync(); - - await Task.Delay(500); - - await server.StopAsync(); - - flow = string.Join(string.Empty, events); - Assert.AreEqual("cdcrd", flow); + await client.PublishAsync("topic1"); + await Task.Delay(100); + Assert.AreEqual(1, receivedMessages.Count); } } [TestMethod] - public async Task Remove_Session() + public async Task Shutdown_Disconnects_Clients_Gracefully() { using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder()); - var clientOptions = new MqttClientOptionsBuilder(); - var c1 = await testEnvironment.ConnectClient(clientOptions); - await Task.Delay(500); - Assert.AreEqual(1, (await server.GetClientsAsync()).Count); + var disconnectCalled = 0; - await c1.DisconnectAsync(); - await Task.Delay(500); + var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder()); + c1.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(e => disconnectCalled++); - Assert.AreEqual(0, (await server.GetClientsAsync()).Count); + await Task.Delay(100); + + await server.StopAsync(); + + await Task.Delay(100); + + Assert.AreEqual(1, disconnectCalled); } } @@ -964,89 +909,176 @@ namespace MQTTnet.Tests.Server } [TestMethod] - public async Task Do_Not_Send_Retained_Messages_For_Denied_Subscription() + [DataRow("", null)] + [DataRow("", "")] + [DataRow(null, null)] + public async Task Use_Admissible_Credentials(string username, string password) { using (var testEnvironment = CreateTestEnvironment()) { - var server = await testEnvironment.StartServer(); + await testEnvironment.StartServer(); - server.InterceptingSubscriptionAsync += e => - { - // This should lead to no subscriptions for "n" at all. So also no sending of retained messages. - if (e.TopicFilter.Topic == "n") - { - e.Response.ReasonCode = MqttSubscribeReasonCode.UnspecifiedError; - } + var client = testEnvironment.CreateClient(); - return Task.CompletedTask; - }; + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost", testEnvironment.ServerPort) + .WithCredentials(username, password) + .Build(); - // Prepare some retained messages. - var client1 = await testEnvironment.ConnectClient(); - await client1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("y").WithPayload("x").WithRetainFlag().Build()); - await client1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("n").WithPayload("x").WithRetainFlag().Build()); - await client1.DisconnectAsync(); + var connectResult = await client.ConnectAsync(clientOptions); - await Task.Delay(500); - - // Subscribe to all retained message types. - // It is important to do this in a range of filters to ensure that a subscription is not "hidden". - var client2 = await testEnvironment.ConnectClient(); - - var buffer = new StringBuilder(); - - client2.ApplicationMessageReceivedAsync += e => - { - lock (buffer) - { - buffer.Append(e.ApplicationMessage.Topic); - } - - return Task.CompletedTask; - }; - - await client2.SubscribeAsync(new MqttTopicFilter {Topic = "y"}, new MqttTopicFilter {Topic = "n"}); - - await Task.Delay(500); - - Assert.AreEqual("y", buffer.ToString()); + Assert.IsFalse(connectResult.IsSessionPresent); + Assert.IsTrue(client.IsConnected); } } [TestMethod] - public async Task Collect_Messages_In_Disconnected_Session() + public async Task Use_Clean_Session() { using (var testEnvironment = CreateTestEnvironment()) { - var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithPersistentSessions()); + await testEnvironment.StartServer(); - // Create the session including the subscription. - var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a").WithCleanSession(false)); - await client1.SubscribeAsync("x"); - await client1.DisconnectAsync(); - await Task.Delay(500); + var client = testEnvironment.CreateClient(); + var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build()); - var clientStatus = await server.GetClientsAsync(); - Assert.AreEqual(0, clientStatus.Count); - - var client2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("b").WithCleanSession(false)); - await client2.PublishAsync("x", "1"); - await client2.PublishAsync("x", "2"); - await client2.PublishAsync("x", "3"); - await client2.DisconnectAsync(); - - await Task.Delay(500); - - clientStatus = await server.GetClientsAsync(); - var sessionStatus = await server.GetSessionsAsync(); - - Assert.AreEqual(0, clientStatus.Count); - Assert.AreEqual(2, sessionStatus.Count); - - Assert.AreEqual(3, sessionStatus.First(s => s.ClientId == client1.Options.ClientId).PendingApplicationMessagesCount); + Assert.IsFalse(connectResult.IsSessionPresent); } } + [TestMethod] + public async Task Use_Empty_Client_ID() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var client = testEnvironment.CreateClient(); + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost", testEnvironment.ServerPort) + .WithClientId(string.Empty) + .Build(); + + var connectResult = await client.ConnectAsync(clientOptions); + + Assert.IsFalse(connectResult.IsSessionPresent); + Assert.IsTrue(client.IsConnected); + } + } + + [TestMethod] + public async Task Will_Message_Do_Not_Send_On_Clean_Disconnect() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var receivedMessagesCount = 0; + + await testEnvironment.StartServer(); + + var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").Build(); + + var clientOptions = new MqttClientOptionsBuilder().WithWillMessage(willMessage); + + var c1 = await testEnvironment.ConnectClient(); + + c1.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return Task.CompletedTask; + }; + + await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); + + var c2 = await testEnvironment.ConnectClient(clientOptions); + await c2.DisconnectAsync().ConfigureAwait(false); + + await Task.Delay(1000); + + Assert.AreEqual(0, receivedMessagesCount); + } + } + + [TestMethod] + public async Task Will_Message_Do_Not_Send_On_Takeover() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var receivedMessagesCount = 0; + + await testEnvironment.StartServer(); + + // C1 will receive the last will! + var c1 = await testEnvironment.ConnectClient(); + c1.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return Task.CompletedTask; + }; + + await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); + + // C2 has the last will defined. + var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").Build(); + + var clientOptions = new MqttClientOptionsBuilder() + .WithWillMessage(willMessage) + .WithClientId("WillOwner"); + + var c2 = await testEnvironment.ConnectClient(clientOptions); + + // C3 will do the connection takeover. + var c3 = await testEnvironment.ConnectClient(clientOptions); + + await Task.Delay(1000); + + Assert.AreEqual(0, receivedMessagesCount); + } + } + + [TestMethod] + public async Task Will_Message_Send() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build(); + + var clientOptions = new MqttClientOptionsBuilder().WithWillMessage(willMessage); + + var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder()); + + var receivedMessagesCount = 0; + c1.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return Task.CompletedTask; + }; + + await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); + + var c2 = await testEnvironment.ConnectClient(clientOptions); + c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent. + + await Task.Delay(1000); + + Assert.AreEqual(1, receivedMessagesCount); + } + } + + void ConnectionValidationHandler(ValidatingConnectionEventArgs eventArgs) + { + if (_connected.ContainsKey(eventArgs.ClientId)) + { + eventArgs.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; + return; + } + + _connected[eventArgs.ClientId] = true; + eventArgs.ReasonCode = MqttConnectReasonCode.Success; + } + async Task TestPublishAsync( string topic, MqttQualityOfServiceLevel qualityOfServiceLevel, @@ -1075,95 +1107,5 @@ namespace MQTTnet.Tests.Server Assert.AreEqual(expectedReceivedMessagesCount, c1MessageHandler.ReceivedEventArgs.Count); } } - - [TestMethod] - public async Task Intercept_Undelivered() - { - using (var testEnvironment = CreateTestEnvironment()) - { - var undeliverd = string.Empty; - - var server = await testEnvironment.StartServer(); - server.ApplicationMessageNotConsumedAsync += e => - { - undeliverd = e.ApplicationMessage.Topic; - return Task.CompletedTask; - }; - - var client = await testEnvironment.ConnectClient(); - - await client.SubscribeAsync("b"); - - await client.PublishAsync("a", null, MqttQualityOfServiceLevel.ExactlyOnce); - - await Task.Delay(500); - - Assert.AreEqual("a", undeliverd); - } - } - - [TestMethod] - public async Task Intercept_ClientMessageQueue() - { - throw new NotImplementedException(); - - // using (var testEnvironment = CreateTestEnvironment()) - // { - // await testEnvironment.StartServer(new MqttServerOptionsBuilder() - // .WithClientMessageQueueInterceptor(c => c.ApplicationMessage.Topic = "a")); - // - // var topicAReceived = false; - // var topicBReceived = false; - // - // var client = await testEnvironment.ConnectClient(); - // client.UseApplicationMessageReceivedHandler(c => - // { - // if (c.ApplicationMessage.Topic == "a") - // { - // topicAReceived = true; - // } - // else if (c.ApplicationMessage.Topic == "b") - // { - // topicBReceived = true; - // } - // }); - // - // await client.SubscribeAsync("a"); - // await client.SubscribeAsync("b"); - // - // await client.PublishAsync("b"); - // - // await Task.Delay(500); - // - // Assert.IsTrue(topicAReceived); - // Assert.IsFalse(topicBReceived); - // } - } - - [TestMethod] - public async Task Intercept_ClientMessageQueue_Different_QoS_Of_Subscription_And_Message() - { - throw new NotImplementedException(); - - const string topic = "a"; - - // using (var testEnvironment = CreateTestEnvironment()) - // { - // await testEnvironment.StartServer(new MqttServerOptionsBuilder() - // .WithClientMessageQueueInterceptor(c => { })); // Interceptor does nothing but has to be present. - // - // bool receivedMessage = false; - // var client = await testEnvironment.ConnectClient(); - // client.UseApplicationMessageReceivedHandler(c => { receivedMessage = true; }); - // - // await client.SubscribeAsync(topic, MqttQualityOfServiceLevel.AtLeastOnce); - // - // await client.PublishAsync(new MqttApplicationMessage {Topic = topic, QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce}); - // - // await Task.Delay(500); - // - // Assert.IsTrue(receivedMessage); - // } - } } } \ No newline at end of file diff --git a/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs index e6627229..04aa1b8d 100644 --- a/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs @@ -32,7 +32,7 @@ namespace MQTTnet.Tests.Server retainedMessagesManager, new MqttClientSessionsManager(options, retainedMessagesManager, eventContainer, logger)); - _subscriptionsManager = new MqttClientSubscriptionsManager(session, new MqttServerOptions(), new MqttServerEventContainer(), + _subscriptionsManager = new MqttClientSubscriptionsManager(session, new MqttServerEventContainer(), new MqttRetainedMessagesManager(new MqttServerEventContainer(), new MqttNetNullLogger())); } diff --git a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs index 5c4bfbb4..ca07aff0 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs @@ -6,6 +6,7 @@ using MQTTnet.Client; using MQTTnet.Protocol; using MQTTnet.Server; using MQTTnet.Tests.Mockups; +using IMqttClient = MQTTnet.Client.IMqttClient; namespace MQTTnet.Tests.Server { @@ -124,18 +125,16 @@ namespace MQTTnet.Tests.Server { // Arrange client and server. var server = await testEnvironment.StartServer(o => o.WithPersistentSessions(false)); - var client = await testEnvironment.ConnectClient(); - - // Arrange session status tracking. - var status = await server.GetSessionsAsync(); - var clientStatus = status[0]; - + var deletedEventFired = false; - clientStatus.Deleted += (_, __) => + server.SessionDeletedAsync += e => { deletedEventFired = true; + return Task.CompletedTask; }; + var client = await testEnvironment.ConnectClient(); + // Act: Disconnect the client -> Event must be fired. await client.DisconnectAsync(); diff --git a/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs index d8ed4233..ba6f918a 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs @@ -1,4 +1,4 @@ -using System.Linq; +using System.Linq; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; @@ -29,8 +29,8 @@ namespace MQTTnet.Tests.Server Assert.AreEqual(2, clientStatus.Count); Assert.AreEqual(2, sessionStatus.Count); - Assert.IsTrue(clientStatus.Any(s => s.ClientId == c1.Options.ClientId)); - Assert.IsTrue(clientStatus.Any(s => s.ClientId == c2.Options.ClientId)); + Assert.IsTrue(clientStatus.Any(s => s.Id == c1.Options.ClientId)); + Assert.IsTrue(clientStatus.Any(s => s.Id == c2.Options.ClientId)); await c1.DisconnectAsync(); await c2.DisconnectAsync(); @@ -59,7 +59,7 @@ namespace MQTTnet.Tests.Server var clientStatus = await server.GetClientsAsync(); Assert.AreEqual(1, clientStatus.Count); - Assert.IsTrue(clientStatus.Any(s => s.ClientId == c1.Options.ClientId)); + Assert.IsTrue(clientStatus.Any(s => s.Id == c1.Options.ClientId)); await clientStatus.First().DisconnectAsync(); diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs index 07a5d69c..4edd5183 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs @@ -9,6 +9,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.FileProviders; using MQTTnet.AspNetCore; using MQTTnet.AspNetCore.Extensions; +using MQTTnet.Server; namespace MQTTnet.TestApp.AspNetCore2 { @@ -55,7 +56,12 @@ namespace MQTTnet.TestApp.AspNetCore2 { try { - await server.PublishAsync("server", msg.Build()); + await server.InjectApplicationMessage(new MqttServer.MqttInjectedApplicationMessage + { + SenderClientId = "server", + ApplicationMessage = msg.Build() + }); + msg.WithPayload($"Mqtt hosted on {frameworkName} is still awesome at {DateTime.Now}"); } catch (Exception e) diff --git a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs index 3f201047..2010a2c9 100644 --- a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs @@ -7,6 +7,7 @@ using System.Threading.Tasks; using MQTTnet.Client; using MQTTnet.Protocol; using MQTTnet.Server; +using IMqttClient = MQTTnet.Client.IMqttClient; namespace MQTTnet.TestApp.NetCore { diff --git a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs index 70620ca5..8deb2a4b 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs @@ -86,7 +86,7 @@ namespace MQTTnet.TestApp.NetCore retainedMessages = new List(); } - e.StoredRetainedMessages = retainedMessages; + e.LoadedRetainedMessages = retainedMessages; return Task.FromResult(0); };