YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-18 02:13:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

Adjust namespaces.

Browse Source
This commit is contained in:
Christian
2021-11-20 10:48:18 +01:00
parent 28310c69bb
commit c59969067c
45 changed files with 1482 additions and 1335 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Build/MQTTnet.nuspec
+5
View File
@@ -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
</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2021</copyright>
MQTTnet.sln.DotSettings
+235
View File
@@ -1,9 +1,244 @@
<wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpCodeStyle/BRACES_FOR_FOR/@EntryValue">Required</s:String>
<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpCodeStyle/BRACES_FOR_FOREACH/@EntryValue">Required</s:String>
<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpCodeStyle/BRACES_FOR_IFELSE/@EntryValue">Required</s:String>
<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpCodeStyle/BRACES_FOR_WHILE/@EntryValue">Required</s:String>
<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_ACCESSORHOLDER_ATTRIBUTE_ON_SAME_LINE_EX/@EntryValue">NEVER</s:String>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_METHOD_CALL_PARENTHESES/@EntryValue">False</s:Boolean>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_METHOD_PARENTHESES/@EntryValue">False</s:Boolean>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_PARENTHESES/@EntryValue">False</s:Boolean>
<s:Int64 x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/WRAP_LIMIT/@EntryValue">180</s:Int64>
<s:String x:Key="/Default/CodeStyle/CSharpFileLayoutPatterns/Pattern/@EntryValue">&lt;Patterns xmlns="urn:schemas-jetbrains-com:member-reordering-patterns"&gt;
&lt;TypePattern DisplayName="Non-reorderable types"&gt;
&lt;TypePattern.Match&gt;
&lt;Or&gt;
&lt;And&gt;
&lt;Kind Is="Interface" /&gt;
&lt;Or&gt;
&lt;HasAttribute Name="System.Runtime.InteropServices.InterfaceTypeAttribute" /&gt;
&lt;HasAttribute Name="System.Runtime.InteropServices.ComImport" /&gt;
&lt;/Or&gt;
&lt;/And&gt;
&lt;Kind Is="Struct" /&gt;
&lt;HasAttribute Name="JetBrains.Annotations.NoReorderAttribute" /&gt;
&lt;HasAttribute Name="JetBrains.Annotations.NoReorder" /&gt;
&lt;/Or&gt;
&lt;/TypePattern.Match&gt;
&lt;/TypePattern&gt;
&lt;TypePattern DisplayName="xUnit.net Test Classes" RemoveRegions="All"&gt;
&lt;TypePattern.Match&gt;
&lt;And&gt;
&lt;Kind Is="Class" /&gt;
&lt;HasMember&gt;
&lt;And&gt;
&lt;Kind Is="Method" /&gt;
&lt;HasAttribute Name="Xunit.FactAttribute" Inherited="True" /&gt;
&lt;HasAttribute Name="Xunit.TheoryAttribute" Inherited="True" /&gt;
&lt;/And&gt;
&lt;/HasMember&gt;
&lt;/And&gt;
&lt;/TypePattern.Match&gt;
&lt;Entry DisplayName="Setup/Teardown Methods"&gt;
&lt;Entry.Match&gt;
&lt;Or&gt;
&lt;Kind Is="Constructor" /&gt;
&lt;And&gt;
&lt;Kind Is="Method" /&gt;
&lt;ImplementsInterface Name="System.IDisposable" /&gt;
&lt;/And&gt;
&lt;/Or&gt;
&lt;/Entry.Match&gt;
&lt;Entry.SortBy&gt;
&lt;Kind&gt;
&lt;Kind.Order&gt;
&lt;DeclarationKind&gt;Constructor&lt;/DeclarationKind&gt;
&lt;/Kind.Order&gt;
&lt;/Kind&gt;
&lt;/Entry.SortBy&gt;
&lt;/Entry&gt;
&lt;Entry DisplayName="All other members" /&gt;
&lt;Entry DisplayName="Test Methods" Priority="100"&gt;
&lt;Entry.Match&gt;
&lt;And&gt;
&lt;Kind Is="Method" /&gt;
&lt;HasAttribute Name="Xunit.FactAttribute" Inherited="false" /&gt;
&lt;HasAttribute Name="Xunit.TheoryAttribute" Inherited="false" /&gt;
&lt;/And&gt;
&lt;/Entry.Match&gt;
&lt;Entry.SortBy&gt;
&lt;Name /&gt;
&lt;/Entry.SortBy&gt;
&lt;/Entry&gt;
&lt;/TypePattern&gt;
&lt;TypePattern DisplayName="NUnit Test Fixtures" RemoveRegions="All"&gt;
&lt;TypePattern.Match&gt;
&lt;And&gt;
&lt;Kind Is="Class" /&gt;
&lt;Or&gt;
&lt;HasAttribute Name="NUnit.Framework.TestFixtureAttribute" Inherited="true" /&gt;
&lt;HasAttribute Name="NUnit.Framework.TestFixtureSourceAttribute" Inherited="true" /&gt;
&lt;HasMember&gt;
&lt;And&gt;
&lt;Kind Is="Method" /&gt;
&lt;HasAttribute Name="NUnit.Framework.TestAttribute" Inherited="false" /&gt;
&lt;HasAttribute Name="NUnit.Framework.TestCaseAttribute" Inherited="false" /&gt;
&lt;HasAttribute Name="NUnit.Framework.TestCaseSourceAttribute" Inherited="false" /&gt;
&lt;/And&gt;
&lt;/HasMember&gt;
&lt;/Or&gt;
&lt;/And&gt;
&lt;/TypePattern.Match&gt;
&lt;Entry DisplayName="Setup/Teardown Methods"&gt;
&lt;Entry.Match&gt;
&lt;And&gt;
&lt;Kind Is="Method" /&gt;
&lt;Or&gt;
&lt;HasAttribute Name="NUnit.Framework.SetUpAttribute" Inherited="true" /&gt;
&lt;HasAttribute Name="NUnit.Framework.TearDownAttribute" Inherited="true" /&gt;
&lt;HasAttribute Name="NUnit.Framework.TestFixtureSetUpAttribute" Inherited="true" /&gt;
&lt;HasAttribute Name="NUnit.Framework.TestFixtureTearDownAttribute" Inherited="true" /&gt;
&lt;HasAttribute Name="NUnit.Framework.OneTimeSetUpAttribute" Inherited="true" /&gt;
&lt;HasAttribute Name="NUnit.Framework.OneTimeTearDownAttribute" Inherited="true" /&gt;
&lt;/Or&gt;
&lt;/And&gt;
&lt;/Entry.Match&gt;
&lt;/Entry&gt;
&lt;Entry DisplayName="All other members" /&gt;
&lt;Entry DisplayName="Test Methods" Priority="100"&gt;
&lt;Entry.Match&gt;
&lt;And&gt;
&lt;Kind Is="Method" /&gt;
&lt;HasAttribute Name="NUnit.Framework.TestAttribute" Inherited="false" /&gt;
&lt;HasAttribute Name="NUnit.Framework.TestCaseAttribute" Inherited="false" /&gt;
&lt;HasAttribute Name="NUnit.Framework.TestCaseSourceAttribute" Inherited="false" /&gt;
&lt;/And&gt;
&lt;/Entry.Match&gt;
&lt;Entry.SortBy&gt;
&lt;Name /&gt;
&lt;/Entry.SortBy&gt;
&lt;/Entry&gt;
&lt;/TypePattern&gt;
&lt;TypePattern DisplayName="Default Pattern"&gt;
&lt;Entry DisplayName="Public Delegates" Priority="100"&gt;
&lt;Entry.Match&gt;
&lt;And&gt;
&lt;Access Is="Public" /&gt;
&lt;Kind Is="Delegate" /&gt;
&lt;/And&gt;
&lt;/Entry.Match&gt;
&lt;Entry.SortBy&gt;
&lt;Name /&gt;
&lt;/Entry.SortBy&gt;
&lt;/Entry&gt;
&lt;Entry DisplayName="Public Enums" Priority="100"&gt;
&lt;Entry.Match&gt;
&lt;And&gt;
&lt;Access Is="Public" /&gt;
&lt;Kind Is="Enum" /&gt;
&lt;/And&gt;
&lt;/Entry.Match&gt;
&lt;Entry.SortBy&gt;
&lt;Name /&gt;
&lt;/Entry.SortBy&gt;
&lt;/Entry&gt;
&lt;Entry DisplayName="Static Fields and Constants"&gt;
&lt;Entry.Match&gt;
&lt;Or&gt;
&lt;Kind Is="Constant" /&gt;
&lt;And&gt;
&lt;Kind Is="Field" /&gt;
&lt;Static /&gt;
&lt;/And&gt;
&lt;/Or&gt;
&lt;/Entry.Match&gt;
&lt;Entry.SortBy&gt;
&lt;Kind&gt;
&lt;Kind.Order&gt;
&lt;DeclarationKind&gt;Constant&lt;/DeclarationKind&gt;
&lt;DeclarationKind&gt;Field&lt;/DeclarationKind&gt;
&lt;/Kind.Order&gt;
&lt;/Kind&gt;
&lt;/Entry.SortBy&gt;
&lt;/Entry&gt;
&lt;Entry DisplayName="Fields"&gt;
&lt;Entry.Match&gt;
&lt;And&gt;
&lt;Kind Is="Field" /&gt;
&lt;Not&gt;
&lt;Static /&gt;
&lt;/Not&gt;
&lt;/And&gt;
&lt;/Entry.Match&gt;
&lt;Entry.SortBy&gt;
&lt;Readonly /&gt;
&lt;Name /&gt;
&lt;/Entry.SortBy&gt;
&lt;/Entry&gt;
&lt;Entry DisplayName="Constructors"&gt;
&lt;Entry.Match&gt;
&lt;Kind Is="Constructor" /&gt;
&lt;/Entry.Match&gt;
&lt;Entry.SortBy&gt;
&lt;Static/&gt;
&lt;/Entry.SortBy&gt;
&lt;/Entry&gt;
&lt;Entry DisplayName="Events"&gt;
&lt;Entry.Match&gt;
&lt;Kind Is="Event" /&gt;
&lt;/Entry.Match&gt;
&lt;Entry.SortBy&gt;
&lt;Name Direction="Ascending" /&gt;
&lt;/Entry.SortBy&gt;
&lt;/Entry&gt;
&lt;Entry DisplayName="Properties, Indexers"&gt;
&lt;Entry.Match&gt;
&lt;Or&gt;
&lt;Kind Is="Property" /&gt;
&lt;Kind Is="Indexer" /&gt;
&lt;/Or&gt;
&lt;/Entry.Match&gt;
&lt;/Entry&gt;
&lt;Entry DisplayName="All other members"&gt;
&lt;Entry.SortBy&gt;
&lt;Access Order="public internal protected private" /&gt;
&lt;Name Direction="Ascending" /&gt;
&lt;/Entry.SortBy&gt;
&lt;/Entry&gt;
&lt;Entry DisplayName="Nested Types"&gt;
&lt;Entry.Match&gt;
&lt;Kind Is="Type" /&gt;
&lt;/Entry.Match&gt;
&lt;/Entry&gt;
&lt;/TypePattern&gt;
&lt;/Patterns&gt;
</s:String>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpAttributeForSingleLineMethodUpgrade/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs
+1
View File
@@ -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
{
Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs
+11 -11
View File
@@ -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;
Source/MQTTnet/Formatter/MqttDisconnectPacketFactory.cs
+8
View File
@@ -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();
Source/MQTTnet/Internal/AsyncEvent.cs
+64 -12
View File
@@ -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<TEventArgs> where TEventArgs : EventArgs
{
readonly List<Func<TEventArgs, Task>> _handlers = new List<Func<TEventArgs, Task>>();
readonly List<AsyncEventInvocator<TEventArgs>> _handlers = new List<AsyncEventInvocator<TEventArgs>>();
public void AddHandler(Func<TEventArgs, Task> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handlers.Add(handler);
}
if (handler == null)
{
throw new ArgumentNullException(nameof(handler));
}
public void RemoveHandler(Func<TEventArgs, Task> handler)
_handlers.Add(new AsyncEventInvocator<TEventArgs>(null, handler));
}
public void AddHandler(Action<TEventArgs> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handlers.Remove(handler);
if (handler == null)
{
throw new ArgumentNullException(nameof(handler));
}
_handlers.Add(new AsyncEventInvocator<TEventArgs>(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<TEventArgs> InvokeAsync(Func<TEventArgs> 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<TEventArgs, Task> handler)
{
if (handler == null)
{
throw new ArgumentNullException(nameof(handler));
}
_handlers.RemoveAll(h => h.WrapsHandler(handler));
}
public void RemoveHandler(Action<TEventArgs> 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)}).");
}
}
}
}
Source/MQTTnet/Internal/AsyncEventInvocator.cs
+44
View File
@@ -0,0 +1,44 @@
using System;
using System.Threading.Tasks;
using MQTTnet.Implementations;
namespace MQTTnet.Internal
{
public sealed class AsyncEventInvocator<TEventArgs>
{
readonly Action<TEventArgs> _handler;
readonly Func<TEventArgs, Task> _asyncHandler;
public AsyncEventInvocator(Action<TEventArgs> handler, Func<TEventArgs, Task> asyncHandler)
{
_handler = handler;
_asyncHandler = asyncHandler;
}
public bool WrapsHandler(Action<TEventArgs> handler)
{
return ReferenceEquals(_handler, handler);
}
public bool WrapsHandler(Func<TEventArgs, Task> 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();
}
}
}
Source/MQTTnet/Internal/MqttPacketBus.cs
+57 -45
View File
@@ -9,9 +9,7 @@ namespace MQTTnet.Internal
{
public sealed class MqttPacketBus : IDisposable
{
readonly object _syncRoot = new object();
readonly LinkedList<MqttPacketBusItem>[] _partitions = new LinkedList<MqttPacketBusItem>[]
readonly LinkedList<MqttPacketBusItem>[] _partitions =
{
new LinkedList<MqttPacketBusItem>(),
new LinkedList<MqttPacketBusItem>(),
@@ -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<MqttBasePacket> 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<MqttPacketBusItem> DequeueAsync(CancellationToken cancellationToken)
public async Task<MqttPacketBusItem> 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<MqttBasePacket> 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;
}
}
}
}
Source/MQTTnet/MQTTnet.csproj.DotSettings
+1
View File
@@ -12,6 +12,7 @@
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=diagnostics_005Cruntime/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cevents/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cinternal/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cmanagement/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Coptions/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstatus/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstorage/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
Source/MQTTnet/MqttFactory.cs
+1
View File
@@ -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
Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs
+2 -1
View File
@@ -1,4 +1,5 @@
using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;
@@ -17,7 +18,7 @@ namespace MQTTnet.Server
/// <summary>
/// Gets or sets a key/value collection that can be used to share data within the scope of this session.
/// </summary>
public IDictionary<object, object> SessionItems { get; internal set; }
public IDictionary SessionItems { get; internal set; }
/// <summary>
/// Gets the response which will be sent to the client via the PUBACK etc. packets.
Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs
+3 -3
View File
@@ -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
/// <summary>
/// Gets or sets a key/value collection that can be used to share data within the scope of this session.
/// </summary>
public IDictionary<object, object> SessionItems { get; internal set; }
public IDictionary SessionItems { get; internal set; }
/// <summary>
/// Gets or sets whether the broker should create an internal subscription for the client.
@@ -39,7 +39,7 @@ namespace MQTTnet.Server
/// <summary>
/// Gets the current client session.
/// </summary>
public IMqttSessionStatus Session { get; internal set; }
public MqttSessionStatus Session { get; internal set; }
/// <summary>
/// Gets the response which will be sent to the client via the SUBACK packet.
Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs
+2 -1
View File
@@ -1,4 +1,5 @@
using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;
@@ -22,7 +23,7 @@ namespace MQTTnet.Server
/// <summary>
/// Gets or sets a key/value collection that can be used to share data within the scope of this session.
/// </summary>
public IDictionary<object, object> SessionItems { get; internal set; }
public IDictionary SessionItems { get; internal set; }
/// <summary>
/// Gets the response which will be sent to the client via the UNSUBACK pocket.
Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs
+1 -1
View File
@@ -5,6 +5,6 @@ namespace MQTTnet.Server
{
public sealed class LoadingRetainedMessagesEventArgs : EventArgs
{
public List<MqttApplicationMessage> StoredRetainedMessages { get; set; } = new List<MqttApplicationMessage>();
public List<MqttApplicationMessage> LoadedRetainedMessages { get; set; } = new List<MqttApplicationMessage>();
}
}
Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs
+1 -1
View File
@@ -26,7 +26,7 @@ namespace MQTTnet.Server
public List<MqttPublishPacket> PublishPackets { get; } = new List<MqttPublishPacket>();
IDictionary<object, object> Items { get; set; }
public IDictionary<object, object> Items { get; set; }
/// <summary>
/// Gets the last will message.
Source/MQTTnet/Server/Events/SessionDeletedEventArgs.cs
+12
View File
@@ -0,0 +1,12 @@
using System;
namespace MQTTnet.Server
{
public sealed class SessionDeletedEventArgs : EventArgs
{
/// <summary>
/// Gets the ID of the session.
/// </summary>
public string Id { get; internal set; }
}
}
Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs
+2 -1
View File
@@ -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
/// <summary>
/// Gets or sets a key/value collection that can be used to share data within the scope of this session.
/// </summary>
public IDictionary<object, object> SessionItems { get; internal set; }
public IDictionary SessionItems { get; internal set; }
/// <summary>
/// This is used for MQTTv3 only.
Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs
+5 -4
View File
@@ -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<object, object> _sessionItems;
readonly IDictionary _sessionItems;
public MqttApplicationMessageInterceptorInvoker(MqttServerEventContainer eventContainer, string clientId, IDictionary<object, object> 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;
}
Source/MQTTnet/Server/Internal/MqttClient.cs
+7 -13
View File
@@ -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);
}
}
}
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+38 -92
View File
@@ -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<string, MqttClient> _clients = new Dictionary<string, MqttClient>(4096);
readonly Dictionary<string, MqttSession> _sessions = new Dictionary<string, MqttSession>(4096);
readonly IDictionary<object, object> _serverSessionItems = new ConcurrentDictionary<object, object>();
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<MqttClient> GetConnections()
public List<MqttClient> GetClients()
{
lock (_clients)
{
return _clients.Values.ToList();
}
}
public Task<IList<IMqttClientStatus>> GetClientStatusAsync()
public MqttClient GetClient(string id)
{
var result = new List<IMqttClientStatus>();
lock (_clients)
{
if (!_clients.TryGetValue(id, out var client))
{
throw new InvalidOperationException($"Client with ID '{id}' not found.");
}
return client;
}
}
public Task<IList<MqttClientStatus>> GetClientStatusAsync()
{
var result = new List<MqttClientStatus>();
lock (_clients)
{
@@ -205,12 +212,12 @@ namespace MQTTnet.Server
}
}
return Task.FromResult((IList<IMqttClientStatus>)result);
return Task.FromResult((IList<MqttClientStatus>)result);
}
public Task<IList<IMqttSessionStatus>> GetSessionStatusAsync()
public Task<IList<MqttSessionStatus>> GetSessionStatusAsync()
{
var result = new List<IMqttSessionStatus>();
var result = new List<MqttSessionStatus>();
lock (_sessions)
{
@@ -221,15 +228,8 @@ namespace MQTTnet.Server
}
}
return Task.FromResult((IList<IMqttSessionStatus>)result);
return Task.FromResult((IList<MqttSessionStatus>)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<MqttTopicFilter> 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<InterceptingMqttClientPublishEventArgs> InterceptApplicationMessageAsync(
// IMqttServerApplicationMessageInterceptor interceptor,
// MqttClientConnection clientConnection,
// MqttApplicationMessage applicationMessage)
// {
// string senderClientId;
// IDictionary<object, object> 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<object, object> sessionItems, bool isPersistent)
MqttSession CreateSession(string clientId, IDictionary sessionItems, bool isPersistent)
{
_logger.Verbose("Created a new session for client '{0}'.", clientId);
Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+12 -21
View File
@@ -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<string, MqttSubscription> _subscriptions = new Dictionary<string, MqttSubscription>(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<string, MqttSubscription> _subscriptions = new ConcurrentDictionary<string, MqttSubscription>();
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<MqttSubscription> 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<MqttQualityOfServiceLevel>();
@@ -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
Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs
+1 -1
View File
@@ -36,7 +36,7 @@ namespace MQTTnet.Server
{
_messages.Clear();
foreach (var retainedMessage in eventArgs.StoredRetainedMessages ?? new List<MqttApplicationMessage>())
foreach (var retainedMessage in eventArgs.LoadedRetainedMessages ?? new List<MqttApplicationMessage>())
{
_messages[retainedMessage.Topic] = retainedMessage;
}
Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs
+2
View File
@@ -23,6 +23,8 @@ namespace MQTTnet.Server
public AsyncEvent<PreparingSessionEventArgs> PreparingSessionEvent { get; } = new AsyncEvent<PreparingSessionEventArgs>();
public AsyncEvent<SessionDeletedEventArgs> SessionDeletedEvent { get; } = new AsyncEvent<SessionDeletedEventArgs>();
public AsyncEvent<ApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent<ApplicationMessageNotConsumedEventArgs>();
public AsyncEvent<RetainedMessageChangedEventArgs> RetainedMessageChangedEvent { get; } = new AsyncEvent<RetainedMessageChangedEventArgs>();
Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs
+7 -7
View File
@@ -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)
{
Source/MQTTnet/Server/Internal/MqttSession.cs
+15 -19
View File
@@ -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<object, object> 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; }
/// <summary>
@@ -54,11 +53,11 @@ namespace MQTTnet.Server
public MqttClientSubscriptionsManager SubscriptionsManager { get; }
public IDictionary<object, object> 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<MqttPacketBusItem> 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.
Source/MQTTnet/Server/MqttApplicationMessageResponseReasonCode.cs
-16
View File
@@ -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
}
}
Source/MQTTnet/Server/MqttClientMessageQueueInterceptorContext.cs
-25
View File
@@ -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;
/// <summary>
/// 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).
/// </summary>
public MqttQualityOfServiceLevel SubscriptionQualityOfServiceLevel { get; set; }
}
}
Source/MQTTnet/Server/MqttServer.cs
+230 -176
View File
@@ -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<IMqttServerAdapter> _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<IMqttServerAdapter> 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<EventArgs, Task> StartedAsync
{
add => _eventContainer.StartedEvent.AddHandler(value);
remove => _eventContainer.StartedEvent.RemoveHandler(value);
}
public event Func<EventArgs, Task> StoppedAsync
{
add => _eventContainer.StoppedEvent.AddHandler(value);
remove => _eventContainer.StoppedEvent.RemoveHandler(value);
}
public event Func<ValidatingConnectionEventArgs, Task> ValidatingConnectionAsync
{
add => _eventContainer.ValidatingConnectionEvent.AddHandler(value);
remove => _eventContainer.ValidatingConnectionEvent.RemoveHandler(value);
}
public event Func<ClientConnectedEventArgs, Task> ClientConnectedAsync
{
add => _eventContainer.ClientConnectedEvent.AddHandler(value);
remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value);
}
public event Func<ClientDisconnectedEventArgs, Task> ClientDisconnectedAsync
{
add => _eventContainer.ClientDisconnectedEvent.AddHandler(value);
remove => _eventContainer.ClientDisconnectedEvent.RemoveHandler(value);
}
public event Func<InterceptingPacketEventArgs, Task> InterceptingInboundPacketAsync
{
add => _eventContainer.InterceptingInboundPacketEvent.AddHandler(value);
remove => _eventContainer.InterceptingInboundPacketEvent.RemoveHandler(value);
}
public event Func<InterceptingPacketEventArgs, Task> InterceptingOutboundPacketAsync
{
add => _eventContainer.InterceptingOutboundPacketEvent.AddHandler(value);
remove => _eventContainer.InterceptingOutboundPacketEvent.RemoveHandler(value);
}
public event Func<InterceptingSubscriptionEventArgs, Task> InterceptingSubscriptionAsync
{
add => _eventContainer.InterceptingSubscriptionEvent.AddHandler(value);
remove => _eventContainer.InterceptingSubscriptionEvent.RemoveHandler(value);
}
public event Func<InterceptingUnsubscriptionEventArgs, Task> InterceptingUnsubscriptionAsync
{
add => _eventContainer.InterceptingUnsubscriptionEvent.AddHandler(value);
remove => _eventContainer.InterceptingUnsubscriptionEvent.RemoveHandler(value);
}
public event Func<InterceptingPublishEventArgs, Task> InterceptingPublishAsync
{
add => _eventContainer.InterceptingPublishEvent.AddHandler(value);
remove => _eventContainer.InterceptingPublishEvent.RemoveHandler(value);
}
public event Func<ClientSubscribedTopicEventArgs, Task> ClientSubscribedTopicAsync
{
add => _eventContainer.ClientSubscribedTopicEvent.AddHandler(value);
remove => _eventContainer.ClientSubscribedTopicEvent.RemoveHandler(value);
}
public event Func<ClientUnsubscribedTopicEventArgs, Task> ClientUnsubscribedTopicAsync
{
add => _eventContainer.ClientUnsubscribedTopicEvent.AddHandler(value);
remove => _eventContainer.ClientUnsubscribedTopicEvent.RemoveHandler(value);
}
public event Func<PreparingSessionEventArgs, Task> PreparingSessionAsync
{
add => _eventContainer.PreparingSessionEvent.AddHandler(value);
remove => _eventContainer.PreparingSessionEvent.RemoveHandler(value);
}
public event Func<ApplicationMessageNotConsumedEventArgs, Task> ApplicationMessageNotConsumedAsync
@@ -123,27 +46,123 @@ namespace MQTTnet.Server
remove => _eventContainer.ApplicationMessageNotConsumedEvent.RemoveHandler(value);
}
public event Func<RetainedMessageChangedEventArgs, Task> RetainedMessageChangedAsync
public event Func<ClientConnectedEventArgs, Task> ClientConnectedAsync
{
add => _eventContainer.RetainedMessageChangedEvent.AddHandler(value);
remove => _eventContainer.RetainedMessageChangedEvent.RemoveHandler(value);
add => _eventContainer.ClientConnectedEvent.AddHandler(value);
remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value);
}
public event Func<ClientDisconnectedEventArgs, Task> ClientDisconnectedAsync
{
add => _eventContainer.ClientDisconnectedEvent.AddHandler(value);
remove => _eventContainer.ClientDisconnectedEvent.RemoveHandler(value);
}
public event Func<ClientSubscribedTopicEventArgs, Task> ClientSubscribedTopicAsync
{
add => _eventContainer.ClientSubscribedTopicEvent.AddHandler(value);
remove => _eventContainer.ClientSubscribedTopicEvent.RemoveHandler(value);
}
public event Func<ClientUnsubscribedTopicEventArgs, Task> ClientUnsubscribedTopicAsync
{
add => _eventContainer.ClientUnsubscribedTopicEvent.AddHandler(value);
remove => _eventContainer.ClientUnsubscribedTopicEvent.RemoveHandler(value);
}
public event Func<InterceptingPacketEventArgs, Task> InterceptingInboundPacketAsync
{
add => _eventContainer.InterceptingInboundPacketEvent.AddHandler(value);
remove => _eventContainer.InterceptingInboundPacketEvent.RemoveHandler(value);
}
public event Func<InterceptingPacketEventArgs, Task> InterceptingOutboundPacketAsync
{
add => _eventContainer.InterceptingOutboundPacketEvent.AddHandler(value);
remove => _eventContainer.InterceptingOutboundPacketEvent.RemoveHandler(value);
}
public event Func<InterceptingPublishEventArgs, Task> InterceptingPublishAsync
{
add => _eventContainer.InterceptingPublishEvent.AddHandler(value);
remove => _eventContainer.InterceptingPublishEvent.RemoveHandler(value);
}
public event Func<InterceptingSubscriptionEventArgs, Task> InterceptingSubscriptionAsync
{
add => _eventContainer.InterceptingSubscriptionEvent.AddHandler(value);
remove => _eventContainer.InterceptingSubscriptionEvent.RemoveHandler(value);
}
public event Func<InterceptingUnsubscriptionEventArgs, Task> InterceptingUnsubscriptionAsync
{
add => _eventContainer.InterceptingUnsubscriptionEvent.AddHandler(value);
remove => _eventContainer.InterceptingUnsubscriptionEvent.RemoveHandler(value);
}
public event Func<LoadingRetainedMessagesEventArgs, Task> LoadingRetainedMessageAsync
{
add => _eventContainer.LoadingRetainedMessagesEvent.AddHandler(value);
remove => _eventContainer.LoadingRetainedMessagesEvent.RemoveHandler(value);
}
public event Func<PreparingSessionEventArgs, Task> PreparingSessionAsync
{
add => _eventContainer.PreparingSessionEvent.AddHandler(value);
remove => _eventContainer.PreparingSessionEvent.RemoveHandler(value);
}
public event Func<RetainedMessageChangedEventArgs, Task> RetainedMessageChangedAsync
{
add => _eventContainer.RetainedMessageChangedEvent.AddHandler(value);
remove => _eventContainer.RetainedMessageChangedEvent.RemoveHandler(value);
}
public event Func<EventArgs, Task> RetainedMessagesClearedAsync
{
add => _eventContainer.RetainedMessagesClearedEvent.AddHandler(value);
remove => _eventContainer.RetainedMessagesClearedEvent.RemoveHandler(value);
}
public event Func<SessionDeletedEventArgs, Task> SessionDeletedAsync
{
add => _eventContainer.SessionDeletedEvent.AddHandler(value);
remove => _eventContainer.SessionDeletedEvent.RemoveHandler(value);
}
public event Func<EventArgs, Task> StartedAsync
{
add => _eventContainer.StartedEvent.AddHandler(value);
remove => _eventContainer.StartedEvent.RemoveHandler(value);
}
public event Func<EventArgs, Task> StoppedAsync
{
add => _eventContainer.StoppedEvent.AddHandler(value);
remove => _eventContainer.StoppedEvent.RemoveHandler(value);
}
public event Func<ValidatingConnectionEventArgs, Task> ValidatingConnectionAsync
{
add => _eventContainer.ValidatingConnectionEvent.AddHandler(value);
remove => _eventContainer.ValidatingConnectionEvent.RemoveHandler(value);
}
public bool IsStarted => _cancellationTokenSource != null;
public Task<IList<IMqttClientStatus>> GetClientsAsync()
/// <summary>
/// Deletes all retained messages.
/// </summary>
/// <returns>A task.</returns>
public Task DeleteRetainedMessagesAsync()
{
ThrowIfDisposed();
ThrowIfNotStarted();
return _retainedMessagesManager?.ClearMessages() ?? PlatformAbstractionLayer.CompletedTask;
}
public Task<IList<MqttClientStatus>> GetClientsAsync()
{
ThrowIfDisposed();
ThrowIfNotStarted();
@@ -151,16 +170,8 @@ namespace MQTTnet.Server
return _clientSessionsManager.GetClientStatusAsync();
}
public Task<IList<IMqttSessionStatus>> GetSessionsAsync()
{
ThrowIfDisposed();
ThrowIfNotStarted();
return _clientSessionsManager.GetSessionStatusAsync();
}
/// <summary>
/// Gets all currently cached retained messages.
/// Gets all currently cached retained messages.
/// </summary>
/// <returns>A list of retained messages.</returns>
public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
@@ -172,80 +183,63 @@ namespace MQTTnet.Server
}
/// <summary>
/// Deletes all retained messages.
/// Disconnects the client with the specified ID.
/// </summary>
/// <param name="id">The ID.</param>
/// <param name="reasonCode">The reason code.</param>
/// <returns>A task.</returns>
public Task DeleteRetainedMessagesAsync()
public Task DisconnectClientAsync(string id, MqttDisconnectReasonCode reasonCode)
{
ThrowIfDisposed();
ThrowIfNotStarted();
return _retainedMessagesManager?.ClearMessages() ?? PlatformAbstractionLayer.CompletedTask;
}
/// <summary>
/// Updates the specified retained message.
/// Setting the payload to null or empty will delete the retained message.
/// </summary>
/// <param name="retainedMessage">The retained message.</param>
/// <returns>A task.</returns>
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<MqttTopicFilter> 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<IList<MqttSessionStatus>> 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<string> 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<MqttClientPublishResult> 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<MqttTopicFilter> 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<string> topicFilters)
{
if (clientId == null)
{
throw new ArgumentNullException(nameof(clientId));
}
if (topicFilters == null)
{
throw new ArgumentNullException(nameof(topicFilters));
}
ThrowIfDisposed();
ThrowIfNotStarted();
return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters);
}
/// <summary>
/// Updates the specified retained message.
/// Setting the payload to null or empty will delete the retained message.
/// </summary>
/// <param name="retainedMessage">The retained message.</param>
/// <returns>A task.</returns>
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.");
}
}
}
}
}
Source/MQTTnet/Server/MqttPendingMessagesOverflowStrategy.cs → Source/MQTTnet/Server/Options/MqttPendingMessagesOverflowStrategy.cs
View File
Source/MQTTnet/Server/PublishResponse.cs
+2 -1
View File
@@ -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; }
Source/MQTTnet/Server/Status/IMqttClientStatus.cs
-45
View File
@@ -1,45 +0,0 @@
using System;
using System.Threading.Tasks;
using MQTTnet.Formatter;
namespace MQTTnet.Server
{
public interface IMqttClientStatus
{
/// <summary>
/// Gets the client identifier.
/// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
/// </summary>
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();
}
}
Source/MQTTnet/Server/Status/IMqttSessionStatus.cs
-55
View File
@@ -1,55 +0,0 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttSessionStatus
{
/// <summary>
/// Fired when this session is being deleted.
/// </summary>
event EventHandler Deleted;
/// <summary>
/// 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.
/// </summary>
string ClientId { get; }
/// <summary>
/// Gets the count of messages which are not yet sent to the client but already queued.
/// </summary>
long PendingApplicationMessagesCount { get; }
/// <summary>
/// This items can be used by the library user in order to store custom information.
/// </summary>
IDictionary<object, object> Items { get; }
/// <summary>
/// Enqueues a new application message.
/// </summary>
/// <param name="applicationMessage">The application message.</param>
/// <returns>A task.</returns>
Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage);
/// <summary>
/// Delivers a new application message.
/// </summary>
/// <param name="applicationMessage">The application message.</param>
/// <returns>A task.</returns>
Task DeliverApplicationMessageAsync(MqttApplicationMessage applicationMessage);
/// <summary>
/// Clears the entire queue with pending application messages.
/// </summary>
/// <returns>A task.</returns>
Task ClearApplicationMessagesQueueAsync();
[Obsolete("Please use _ClearApplicationMessagesQueueAsync_ instead. This will be removed soon.")]
Task ClearPendingApplicationMessagesAsync();
Task DeleteAsync();
}
}
Source/MQTTnet/Server/Status/MqttClientStatus.cs
+21 -21
View File
@@ -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));
}
/// <summary>
/// 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.
/// </summary>
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();
}
}
}
Source/MQTTnet/Server/Status/MqttSessionStatus.cs
+7 -29
View File
@@ -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<object, object> 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();
}
}
}
Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs
+1
View File
@@ -2,6 +2,7 @@
using BenchmarkDotNet.Jobs;
using MQTTnet.Client;
using MQTTnet.Server;
using IMqttClient = MQTTnet.Client.IMqttClient;
namespace MQTTnet.Benchmarks
{
Tests/MQTTnet.Core.Tests/Client/ManagedMqttClient_Tests.cs
+1
View File
@@ -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
{
Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs
+32 -32
View File
@@ -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);
Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs
+1
View File
@@ -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
{
Tests/MQTTnet.Core.Tests/Server/General.cs
+630 -688
View File
File diff suppressed because it is too large Load Diff
Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs
+1 -1
View File
@@ -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()));
}
Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs
+6 -7
View File
@@ -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();
Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs
+4 -4
View File
@@ -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();
Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs
+7 -1
View File
@@ -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)
Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs
+1
View File
@@ -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
{
Tests/MQTTnet.TestApp.NetCore/ServerTest.cs
+1 -1
View File
@@ -86,7 +86,7 @@ namespace MQTTnet.TestApp.NetCore
retainedMessages = new List<MqttApplicationMessage>();
}
e.StoredRetainedMessages = retainedMessages;
e.LoadedRetainedMessages = retainedMessages;
return Task.FromResult(0);
};