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

Add extension for RPC calls; replace locks; Refactored storage in Managed Client

Browse Source
This commit is contained in:
Christian Kratky
2017-12-04 10:21:37 +01:00
parent 0ee3c8e135
commit d12d2648b7
31 changed files with 690 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Frameworks/MQTTnet.AspnetCore/MQTTnet.AspnetCore.csproj
+7 -2
View File
@@ -2,8 +2,13 @@
<PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework>
<AssemblyVersion>2.5.2.0</AssemblyVersion>
<FileVersion>2.5.2.0</FileVersion>
<AssemblyVersion>0.0.0.0</AssemblyVersion>
<FileVersion>0.0.0.0</FileVersion>
<Version>0.0.0.0</Version>
<Product />
<Company />
<Authors />
<PackageId />
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">
Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs
+22 -27
View File
@@ -18,6 +18,8 @@ namespace MQTTnet.Adapter
{
private const uint ErrorOperationAborted = 0x800703E3;
private static readonly byte[] EmptyBody = new byte[0];
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly IMqttNetLogger _logger;
private readonly IMqttChannel _channel;
@@ -89,35 +91,28 @@ namespace MQTTnet.Adapter
MqttBasePacket packet = null;
await ExecuteAndWrapExceptionAsync(async () =>
{
ReceivedMqttPacket receivedMqttPacket = null;
try
ReceivedMqttPacket receivedMqttPacket;
if (timeout > TimeSpan.Zero)
{
if (timeout > TimeSpan.Zero)
{
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
}
else
{
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
}
if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
{
throw new TaskCanceledException();
}
packet = PacketSerializer.Deserialize(receivedMqttPacket);
if (packet == null)
{
throw new MqttProtocolViolationException("Received malformed packet.");
}
_logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
}
finally
else
{
receivedMqttPacket?.Dispose();
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
}
if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
{
throw new TaskCanceledException();
}
packet = PacketSerializer.Deserialize(receivedMqttPacket.Header, receivedMqttPacket.Body);
if (packet == null)
{
throw new MqttProtocolViolationException("Received malformed packet.");
}
_logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
}).ConfigureAwait(false);
return packet;
@@ -133,7 +128,7 @@ namespace MQTTnet.Adapter
if (header.BodyLength == 0)
{
return new ReceivedMqttPacket(header, new MemoryStream(0));
return new ReceivedMqttPacket(header, EmptyBody);
}
var body = new byte[header.BodyLength];
@@ -145,7 +140,7 @@ namespace MQTTnet.Adapter
offset += readBytesCount;
} while (offset < header.BodyLength);
return new ReceivedMqttPacket(header, new MemoryStream(body, 0, body.Length, false, true));
return new ReceivedMqttPacket(header, body);
}
private static async Task ExecuteAndWrapExceptionAsync(Func<Task> action)
Frameworks/MQTTnet.NetStandard/Adapter/ReceivedMqttPacket.cs
+3 -9
View File
@@ -1,12 +1,11 @@
using System;
using System.IO;
using MQTTnet.Packets;
namespace MQTTnet.Adapter
{
public sealed class ReceivedMqttPacket : IDisposable
public class ReceivedMqttPacket
{
public ReceivedMqttPacket(MqttPacketHeader header, MemoryStream body)
public ReceivedMqttPacket(MqttPacketHeader header, byte[] body)
{
Header = header ?? throw new ArgumentNullException(nameof(header));
Body = body ?? throw new ArgumentNullException(nameof(body));
@@ -14,11 +13,6 @@ namespace MQTTnet.Adapter
public MqttPacketHeader Header { get; }
public MemoryStream Body { get; }
public void Dispose()
{
Body?.Dispose();
}
public byte[] Body { get; }
}
}
Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs
+2 -5
View File
@@ -70,7 +70,7 @@ namespace MQTTnet.Implementations
_sslStream = new SslStream(new NetworkStream(_socket, true), false, InternalUserCertificateValidationCallback);
await _sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(_options), SslProtocols.Tls12, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
}
CreateStreams(_socket, _sslStream);
}
@@ -139,10 +139,7 @@ namespace MQTTnet.Implementations
private void CreateStreams(Socket socket, Stream sslStream)
{
var stream = sslStream ?? new NetworkStream(socket);
//cannot use this as default buffering prevents from receiving the first connect message
//need two streams otherwise read and write have to be synchronized
//todo: if branch can be used with min dependency NetStandard1.6
#if NET452 || NET461
SendStream = new BufferedStream(stream, BufferSize);
Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj
+2 -3
View File
@@ -6,8 +6,8 @@
<RootNamespace>MQTTnet</RootNamespace>
<GeneratePackageOnBuild>False</GeneratePackageOnBuild>
<DebugType>Full</DebugType>
<AssemblyVersion>2.5.3.0</AssemblyVersion>
<FileVersion>2.5.3.0</FileVersion>
<AssemblyVersion>0.0.0.0</AssemblyVersion>
<FileVersion>0.0.0.0</FileVersion>
<Version>0.0.0.0</Version>
<Company />
<Product />
@@ -29,7 +29,6 @@
<TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
<DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
<LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x86'" />
Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClient.cs
+37 -20
View File
@@ -13,9 +13,9 @@ namespace MQTTnet.ManagedClient
{
public class ManagedMqttClient : IManagedMqttClient
{
private readonly ManagedMqttClientStorageManager _storageManager = new ManagedMqttClientStorageManager();
private readonly BlockingCollection<MqttApplicationMessage> _messageQueue = new BlockingCollection<MqttApplicationMessage>();
private readonly HashSet<TopicFilter> _subscriptions = new HashSet<TopicFilter>();
private readonly SemaphoreSlim _subscriptionsSemaphore = new SemaphoreSlim(1, 1);
private readonly IMqttClient _mqttClient;
private readonly IMqttNetLogger _logger;
@@ -23,7 +23,9 @@ namespace MQTTnet.ManagedClient
private CancellationTokenSource _connectionCancellationToken;
private CancellationTokenSource _publishingCancellationToken;
private ManagedMqttClientStorageManager _storageManager;
private IManagedMqttClientOptions _options;
private bool _subscriptionsNotPushed;
public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
@@ -55,15 +57,11 @@ namespace MQTTnet.ManagedClient
if (_connectionCancellationToken != null) throw new InvalidOperationException("The managed client is already started.");
_options = options;
await _storageManager.SetStorageAsync(_options.Storage).ConfigureAwait(false);
if (_options.Storage != null)
{
var loadedMessages = await _options.Storage.LoadQueuedMessagesAsync().ConfigureAwait(false);
foreach (var loadedMessage in loadedMessages)
{
_messageQueue.Add(loadedMessage);
}
_storageManager = new ManagedMqttClientStorageManager(_options.Storage);
await _storageManager.LoadQueuedMessagesAsync().ConfigureAwait(false);
}
_connectionCancellationToken = new CancellationTokenSource();
@@ -97,16 +95,21 @@ namespace MQTTnet.ManagedClient
foreach (var applicationMessage in applicationMessages)
{
await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
if (_storageManager != null)
{
await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
}
_messageQueue.Add(applicationMessage);
}
}
public Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
public async Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
{
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
lock (_subscriptions)
await _subscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
try
{
foreach (var topicFilter in topicFilters)
{
@@ -116,13 +119,16 @@ namespace MQTTnet.ManagedClient
}
}
}
return Task.FromResult(0);
finally
{
_subscriptionsSemaphore.Release();
}
}
public Task UnsubscribeAsync(IEnumerable<TopicFilter> topicFilters)
public async Task UnsubscribeAsync(IEnumerable<TopicFilter> topicFilters)
{
lock (_subscriptions)
await _subscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
try
{
foreach (var topicFilter in topicFilters)
{
@@ -132,8 +138,10 @@ namespace MQTTnet.ManagedClient
}
}
}
return Task.FromResult(0);
finally
{
_subscriptionsSemaphore.Release();
}
}
private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
@@ -242,7 +250,11 @@ namespace MQTTnet.ManagedClient
try
{
await _mqttClient.PublishAsync(message).ConfigureAwait(false);
await _storageManager.RemoveAsync(message).ConfigureAwait(false);
if (_storageManager != null)
{
await _storageManager.RemoveAsync(message).ConfigureAwait(false);
}
}
catch (MqttCommunicationException exception)
{
@@ -264,13 +276,18 @@ namespace MQTTnet.ManagedClient
_logger.Info<ManagedMqttClient>(nameof(ManagedMqttClient), "Synchronizing subscriptions");
List<TopicFilter> subscriptions;
lock (_subscriptions)
await _subscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
try
{
subscriptions = _subscriptions.ToList();
_subscriptionsNotPushed = false;
}
finally
{
_subscriptionsSemaphore.Release();
}
if (!_subscriptions.Any())
if (!subscriptions.Any())
{
return;
}
Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClientStorageManager.cs
+12 -20
View File
@@ -1,4 +1,5 @@
using System.Collections.Generic;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
@@ -8,18 +9,19 @@ namespace MQTTnet.ManagedClient
{
private readonly List<MqttApplicationMessage> _applicationMessages = new List<MqttApplicationMessage>();
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private IManagedMqttClientStorage _storage;
private readonly IManagedMqttClientStorage _storage;
public async Task SetStorageAsync(IManagedMqttClientStorage storage)
public ManagedMqttClientStorageManager(IManagedMqttClientStorage storage)
{
await _semaphore.WaitAsync().ConfigureAwait(false);
try
_storage = storage ?? throw new ArgumentNullException(nameof(storage));
}
public async Task LoadQueuedMessagesAsync()
{
var loadedMessages = await _storage.LoadQueuedMessagesAsync().ConfigureAwait(false);
foreach (var loadedMessage in loadedMessages)
{
_storage = storage;
}
finally
{
_semaphore.Release();
_applicationMessages.Add(loadedMessage);
}
}
@@ -28,11 +30,6 @@ namespace MQTTnet.ManagedClient
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
if (_storage == null)
{
return;
}
_applicationMessages.Add(applicationMessage);
await SaveAsync().ConfigureAwait(false);
}
@@ -47,11 +44,6 @@ namespace MQTTnet.ManagedClient
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
if (_storage == null)
{
return;
}
var index = _applicationMessages.IndexOf(applicationMessage);
if (index == -1)
{
Frameworks/MQTTnet.NetStandard/Serializer/IMqttPacketSerializer.cs
+2 -3
View File
@@ -1,6 +1,5 @@
using System;
using System.Collections.Generic;
using MQTTnet.Adapter;
using MQTTnet.Packets;
namespace MQTTnet.Serializer
@@ -9,8 +8,8 @@ namespace MQTTnet.Serializer
{
MqttProtocolVersion ProtocolVersion { get; set; }
IEnumerable<ArraySegment<byte>> Serialize(MqttBasePacket mqttPacket);
ICollection<ArraySegment<byte>> Serialize(MqttBasePacket mqttPacket);
MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket);
MqttBasePacket Deserialize(MqttPacketHeader header, byte[] body);
}
}
Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs
+12 -13
View File
@@ -1,9 +1,9 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -12,15 +12,15 @@ namespace MQTTnet.Serializer
{
public sealed class MqttPacketReader : BinaryReader
{
private readonly ReceivedMqttPacket _receivedMqttPacket;
public MqttPacketReader(ReceivedMqttPacket receivedMqttPacket)
: base(receivedMqttPacket.Body, Encoding.UTF8, true)
private readonly MqttPacketHeader _header;
public MqttPacketReader(MqttPacketHeader header, Stream bodyStream)
: base(bodyStream, Encoding.UTF8, true)
{
_receivedMqttPacket = receivedMqttPacket;
_header = header;
}
public bool EndOfRemainingData => BaseStream.Position == _receivedMqttPacket.Header.BodyLength;
public bool EndOfRemainingData => BaseStream.Position == _header.BodyLength;
public static MqttPacketHeader ReadHeaderFromSource(Stream stream, CancellationToken cancellationToken)
{
@@ -77,7 +77,7 @@ namespace MQTTnet.Serializer
public byte[] ReadRemainingData()
{
return ReadBytes(_receivedMqttPacket.Header.BodyLength - (int)BaseStream.Position);
return ReadBytes(_header.BodyLength - (int)BaseStream.Position);
}
private static int ReadBodyLengthFromSource(Stream stream, CancellationToken cancellationToken)
@@ -87,7 +87,7 @@ namespace MQTTnet.Serializer
var value = 0;
byte encodedByte;
////var readBytes = new List<int>();
var readBytes = new List<byte>();
do
{
if (cancellationToken.IsCancellationRequested)
@@ -101,15 +101,14 @@ namespace MQTTnet.Serializer
throw new MqttCommunicationException("Connection closed while reading remaining length data.");
}
////readBytes.Add(buffer);
encodedByte = (byte)buffer;
readBytes.Add(encodedByte);
value += (byte)(encodedByte & 127) * multiplier;
multiplier *= 128;
if (multiplier > 128 * 128 * 128)
{
//throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)}).");
throw new MqttProtocolViolationException("Remaining length is invalid.");
throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)}).");
}
} while ((encodedByte & 128) != 0);
Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs
+8 -7
View File
@@ -1,5 +1,4 @@
using MQTTnet.Adapter;
using MQTTnet.Exceptions;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using System;
@@ -17,7 +16,7 @@ namespace MQTTnet.Serializer
public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
public IEnumerable<ArraySegment<byte>> Serialize(MqttBasePacket packet)
public ICollection<ArraySegment<byte>> Serialize(MqttBasePacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
@@ -43,13 +42,15 @@ namespace MQTTnet.Serializer
}
}
public MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket)
public MqttBasePacket Deserialize(MqttPacketHeader header, byte[] body)
{
if (receivedMqttPacket == null) throw new ArgumentNullException(nameof(receivedMqttPacket));
if (header == null) throw new ArgumentNullException(nameof(header));
if (body == null) throw new ArgumentNullException(nameof(body));
using (var reader = new MqttPacketReader(receivedMqttPacket))
using (var bodyStream = new MemoryStream(body))
using (var reader = new MqttPacketReader(header, bodyStream))
{
return Deserialize(receivedMqttPacket.Header, reader);
return Deserialize(header, reader);
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs
+27 -16
View File
@@ -14,15 +14,17 @@ namespace MQTTnet.Server
{
public sealed class MqttClientSession
{
private readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch();
private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = new Stopwatch();
private readonly Stopwatch _lastPacketReceivedTracker = Stopwatch.StartNew();
private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = Stopwatch.StartNew();
private readonly MqttClientSubscriptionsManager _subscriptionsManager;
private readonly MqttClientSessionsManager _sessionsManager;
private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
private readonly IMqttServerOptions _options;
private readonly IMqttNetLogger _logger;
private readonly MqttClientSessionsManager _sessionsManager;
private readonly MqttRetainedMessagesManager _retainedMessagesManager;
private readonly MqttClientSubscriptionsManager _subscriptionsManager;
private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
private IMqttChannelAdapter _adapter;
private CancellationTokenSource _cancellationTokenSource;
private MqttApplicationMessage _willMessage;
@@ -30,16 +32,17 @@ namespace MQTTnet.Server
public MqttClientSession(
string clientId,
IMqttServerOptions options,
MqttRetainedMessagesManager retainedMessagesManager,
MqttClientSessionsManager sessionsManager,
IMqttNetLogger logger)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
_retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
_sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
ClientId = clientId;
_options = options;
_subscriptionsManager = new MqttClientSubscriptionsManager(_options);
_pendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
}
@@ -117,11 +120,11 @@ namespace MQTTnet.Server
}
}
public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage)
public async Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
var result = _subscriptionsManager.CheckSubscriptions(applicationMessage);
var result = await _subscriptionsManager.CheckSubscriptionsAsync(applicationMessage);
if (!result.IsSubscribed)
{
return;
@@ -129,6 +132,7 @@ namespace MQTTnet.Server
var publishPacket = applicationMessage.ToPublishPacket();
publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel;
_pendingMessagesQueue.Enqueue(publishPacket);
}
@@ -199,7 +203,7 @@ namespace MQTTnet.Server
if (packet is MqttUnsubscribePacket unsubscribePacket)
{
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, _subscriptionsManager.Unsubscribe(unsubscribePacket));
return HandleIncomingUnsubscribePacketAsync(adapter, unsubscribePacket, cancellationToken);
}
if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
@@ -213,24 +217,31 @@ namespace MQTTnet.Server
private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{
var subscribeResult = _subscriptionsManager.Subscribe(subscribePacket, ClientId);
var subscribeResult = await _subscriptionsManager.SubscribeAsync(subscribePacket, ClientId);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, subscribeResult.ResponsePacket).ConfigureAwait(false);
await EnqueueSubscribedRetainedMessagesAsync(subscribePacket).ConfigureAwait(false);
if (subscribeResult.CloseConnection)
{
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()).ConfigureAwait(false);
await StopAsync().ConfigureAwait(false);
}
await EnqueueSubscribedRetainedMessagesAsync(subscribePacket).ConfigureAwait(false);
}
private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
{
var unsubscribeResult = await _subscriptionsManager.UnsubscribeAsync(unsubscribePacket);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, unsubscribeResult);
}
private async Task EnqueueSubscribedRetainedMessagesAsync(MqttSubscribePacket subscribePacket)
{
var retainedMessages = await _sessionsManager.GetRetainedMessagesAsync(subscribePacket).ConfigureAwait(false);
var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(subscribePacket);
foreach (var publishPacket in retainedMessages)
{
EnqueueApplicationMessage(publishPacket);
await EnqueueApplicationMessageAsync(publishPacket);
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs
+37 -34
View File
@@ -15,23 +15,21 @@ namespace MQTTnet.Server
public sealed class MqttClientSessionsManager
{
private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
private readonly SemaphoreSlim _sessionsSemaphore = new SemaphoreSlim(1, 1);
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly IMqttServerOptions _options;
private readonly MqttServer _server;
private readonly MqttRetainedMessagesManager _retainedMessagesManager;
private readonly IMqttNetLogger _logger;
public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, MqttServer server, IMqttNetLogger logger)
{
_server = server ?? throw new ArgumentNullException(nameof(server));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_options = options ?? throw new ArgumentNullException(nameof(options));
_retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
}
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public async Task RunClientSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
{
var clientId = string.Empty;
@@ -66,11 +64,11 @@ namespace MQTTnet.Server
IsSessionPresent = clientSession.IsExistingSession
}).ConfigureAwait(false);
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(new ConnectedMqttClient
_server.OnClientConnected(new ConnectedMqttClient
{
ClientId = clientId,
ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
}));
});
await clientSession.Session.RunAsync(connectPacket.WillMessage, clientAdapter).ConfigureAwait(false);
}
@@ -89,17 +87,17 @@ namespace MQTTnet.Server
// ignored
}
ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(new ConnectedMqttClient
_server.OnClientDisconnected(new ConnectedMqttClient
{
ClientId = clientId,
ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
}));
});
}
}
public async Task StopAsync()
{
await _sessionsSemaphore.WaitAsync().ConfigureAwait(false);
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
foreach (var session in _sessions)
@@ -111,13 +109,13 @@ namespace MQTTnet.Server
}
finally
{
_sessionsSemaphore.Release();
_semaphore.Release();
}
}
public async Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
{
await _sessionsSemaphore.WaitAsync().ConfigureAwait(false);
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
@@ -130,7 +128,7 @@ namespace MQTTnet.Server
}
finally
{
_sessionsSemaphore.Release();
_semaphore.Release();
}
}
@@ -138,17 +136,7 @@ namespace MQTTnet.Server
{
try
{
if (_options.ApplicationMessageInterceptor != null)
{
var interceptorContext = new MqttApplicationMessageInterceptorContext
{
ApplicationMessage = applicationMessage
};
_options.ApplicationMessageInterceptor(interceptorContext);
applicationMessage = interceptorContext.ApplicationMessage;
}
applicationMessage = InterceptApplicationMessage(applicationMessage);
if (applicationMessage == null)
{
return;
@@ -159,26 +147,41 @@ namespace MQTTnet.Server
await _retainedMessagesManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
}
var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, applicationMessage);
ApplicationMessageReceived?.Invoke(this, eventArgs);
_server.OnApplicationMessageReceived(senderClientSession?.ClientId, applicationMessage);
}
catch (Exception exception)
{
_logger.Error<MqttClientSessionsManager>(exception, "Error while processing application message");
}
lock (_sessions)
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
foreach (var clientSession in _sessions.Values)
{
clientSession.EnqueueApplicationMessage(applicationMessage);
await clientSession.EnqueueApplicationMessageAsync(applicationMessage);
}
}
finally
{
_semaphore.Release();
}
}
public Task<List<MqttApplicationMessage>> GetRetainedMessagesAsync(MqttSubscribePacket subscribePacket)
private MqttApplicationMessage InterceptApplicationMessage(MqttApplicationMessage applicationMessage)
{
return _retainedMessagesManager.GetSubscribedMessagesAsync(subscribePacket);
if (_options.ApplicationMessageInterceptor == null)
{
return applicationMessage;
}
var interceptorContext = new MqttApplicationMessageInterceptorContext
{
ApplicationMessage = applicationMessage
};
_options.ApplicationMessageInterceptor(interceptorContext);
return interceptorContext.ApplicationMessage;
}
private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
@@ -200,7 +203,7 @@ namespace MQTTnet.Server
private async Task<GetOrCreateClientSessionResult> GetOrCreateClientSessionAsync(MqttConnectPacket connectPacket)
{
await _sessionsSemaphore.WaitAsync().ConfigureAwait(false);
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession);
@@ -225,7 +228,7 @@ namespace MQTTnet.Server
{
isExistingSession = false;
clientSession = new MqttClientSession(connectPacket.ClientId, _options, this, _logger);
clientSession = new MqttClientSession(connectPacket.ClientId, _options, _retainedMessagesManager, this, _logger);
_sessions[connectPacket.ClientId] = clientSession;
_logger.Trace<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
@@ -235,7 +238,7 @@ namespace MQTTnet.Server
}
finally
{
_sessionsSemaphore.Release();
_semaphore.Release();
}
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs
+87 -26
View File
@@ -1,5 +1,8 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -7,6 +10,7 @@ namespace MQTTnet.Server
{
public sealed class MqttClientSubscriptionsManager
{
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
private readonly IMqttServerOptions _options;
@@ -15,24 +19,34 @@ namespace MQTTnet.Server
_options = options ?? throw new ArgumentNullException(nameof(options));
}
public MqttClientSubscribeResult Subscribe(MqttSubscribePacket subscribePacket, string clientId)
public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket, string clientId)
{
if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
var responsePacket = subscribePacket.CreateResponse<MqttSubAckPacket>();
var closeConnection = false;
var result = new MqttClientSubscribeResult
{
ResponsePacket = subscribePacket.CreateResponse<MqttSubAckPacket>(),
CloseConnection = false
};
lock (_subscriptions)
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
foreach (var topicFilter in subscribePacket.TopicFilters)
{
var interceptorContext = new MqttSubscriptionInterceptorContext(clientId, topicFilter);
_options.SubscriptionInterceptor?.Invoke(interceptorContext);
responsePacket.SubscribeReturnCodes.Add(interceptorContext.AcceptSubscription ? MqttSubscribeReturnCode.SuccessMaximumQoS1 : MqttSubscribeReturnCode.Failure);
var interceptorContext = InterceptSubscribe(clientId, topicFilter);
if (!interceptorContext.AcceptSubscription)
{
result.ResponsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.Failure);
}
else
{
result.ResponsePacket.SubscribeReturnCodes.Add(ConvertToMaximumQoS(topicFilter.QualityOfServiceLevel));
}
if (interceptorContext.CloseConnection)
{
closeConnection = true;
result.CloseConnection = true;
}
if (interceptorContext.AcceptSubscription)
@@ -41,35 +55,42 @@ namespace MQTTnet.Server
}
}
}
return new MqttClientSubscribeResult
finally
{
ResponsePacket = responsePacket,
CloseConnection = closeConnection
};
_semaphore.Release();
}
return result;
}
public MqttUnsubAckPacket Unsubscribe(MqttUnsubscribePacket unsubscribePacket)
public async Task<MqttUnsubAckPacket> UnsubscribeAsync(MqttUnsubscribePacket unsubscribePacket)
{
if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
lock (_subscriptions)
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
foreach (var topicFilter in unsubscribePacket.TopicFilters)
{
_subscriptions.Remove(topicFilter);
}
}
finally
{
_semaphore.Release();
}
return unsubscribePacket.CreateResponse<MqttUnsubAckPacket>();
}
public CheckSubscriptionsResult CheckSubscriptions(MqttApplicationMessage applicationMessage)
public async Task<CheckSubscriptionsResult> CheckSubscriptionsAsync(MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
lock (_subscriptions)
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
var qosLevels = new HashSet<MqttQualityOfServiceLevel>();
foreach (var subscription in _subscriptions)
{
if (!MqttTopicFilterComparer.IsMatch(applicationMessage.Topic, subscription.Key))
@@ -77,24 +98,64 @@ namespace MQTTnet.Server
continue;
}
var effectiveQos = subscription.Value;
if (applicationMessage.QualityOfServiceLevel < effectiveQos)
{
effectiveQos = applicationMessage.QualityOfServiceLevel;
}
qosLevels.Add(subscription.Value);
}
if (qosLevels.Count == 0)
{
return new CheckSubscriptionsResult
{
IsSubscribed = true,
QualityOfServiceLevel = effectiveQos
IsSubscribed = false
};
}
return CreateSubscriptionResult(applicationMessage, qosLevels);
}
finally
{
_semaphore.Release();
}
}
private MqttSubscriptionInterceptorContext InterceptSubscribe(string clientId, TopicFilter topicFilter)
{
var interceptorContext = new MqttSubscriptionInterceptorContext(clientId, topicFilter);
_options.SubscriptionInterceptor?.Invoke(interceptorContext);
return interceptorContext;
}
private static CheckSubscriptionsResult CreateSubscriptionResult(MqttApplicationMessage applicationMessage, HashSet<MqttQualityOfServiceLevel> subscribedQoSLevels)
{
MqttQualityOfServiceLevel effectiveQoS;
if (subscribedQoSLevels.Contains(applicationMessage.QualityOfServiceLevel))
{
effectiveQoS = applicationMessage.QualityOfServiceLevel;
}
else if (subscribedQoSLevels.Count == 1)
{
effectiveQoS = subscribedQoSLevels.First();
}
else
{
effectiveQoS = subscribedQoSLevels.Max();
}
return new CheckSubscriptionsResult
{
IsSubscribed = false
IsSubscribed = true,
QualityOfServiceLevel = effectiveQoS
};
}
private static MqttSubscribeReturnCode ConvertToMaximumQoS(MqttQualityOfServiceLevel qualityOfServiceLevel)
{
switch (qualityOfServiceLevel)
{
case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS0;
case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS1;
case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS2;
default: return MqttSubscribeReturnCode.Failure;
}
}
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs
+7 -7
View File
@@ -11,7 +11,7 @@ namespace MQTTnet.Server
public sealed class MqttRetainedMessagesManager
{
private readonly Dictionary<string, MqttApplicationMessage> _retainedMessages = new Dictionary<string, MqttApplicationMessage>();
private readonly SemaphoreSlim _gate = new SemaphoreSlim(1, 1);
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly IMqttNetLogger _logger;
private readonly IMqttServerOptions _options;
@@ -28,7 +28,7 @@ namespace MQTTnet.Server
return;
}
await _gate.WaitAsync();
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync();
@@ -45,7 +45,7 @@ namespace MQTTnet.Server
}
finally
{
_gate.Release();
_semaphore.Release();
}
}
@@ -53,7 +53,7 @@ namespace MQTTnet.Server
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
await _gate.WaitAsync().ConfigureAwait(false);
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
await HandleMessageInternalAsync(clientId, applicationMessage);
@@ -64,7 +64,7 @@ namespace MQTTnet.Server
}
finally
{
_gate.Release();
_semaphore.Release();
}
}
@@ -72,7 +72,7 @@ namespace MQTTnet.Server
{
var retainedMessages = new List<MqttApplicationMessage>();
await _gate.WaitAsync().ConfigureAwait(false);
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
foreach (var retainedMessage in _retainedMessages.Values)
@@ -96,7 +96,7 @@ namespace MQTTnet.Server
}
finally
{
_gate.Release();
_semaphore.Release();
}
return retainedMessages;
Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs
+27 -32
View File
@@ -60,11 +60,7 @@ namespace MQTTnet.Server
_cancellationTokenSource = new CancellationTokenSource();
_retainedMessagesManager = new MqttRetainedMessagesManager(_options, _logger);
_clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, _logger);
_clientSessionsManager.ApplicationMessageReceived += OnApplicationMessageReceived;
_clientSessionsManager.ClientConnected += OnClientConnected;
_clientSessionsManager.ClientDisconnected += OnClientDisconnected;
_clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, this, _logger);
await _retainedMessagesManager.LoadMessagesAsync();
@@ -104,40 +100,39 @@ namespace MQTTnet.Server
finally
{
_cancellationTokenSource = null;
_retainedMessagesManager = null;
if (_clientSessionsManager != null)
{
_clientSessionsManager.ApplicationMessageReceived -= OnApplicationMessageReceived;
_clientSessionsManager.ClientConnected -= OnClientConnected;
_clientSessionsManager.ClientDisconnected -= OnClientDisconnected;
}
_clientSessionsManager = null;
}
}
internal void OnClientConnected(ConnectedMqttClient client)
{
if (client == null) throw new ArgumentNullException(nameof(client));
_logger.Info<MqttServer>("Client '{0}': Connected.", client.ClientId);
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(client));
}
internal void OnClientDisconnected(ConnectedMqttClient client)
{
if (client == null) throw new ArgumentNullException(nameof(client));
_logger.Info<MqttServer>("Client '{0}': Disconnected.", client.ClientId);
ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client));
}
internal void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(clientId, applicationMessage));
}
private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
{
eventArgs.SessionTask = Task.Run(async () => await _clientSessionsManager.RunClientSessionAsync(eventArgs.Client, _cancellationTokenSource.Token), _cancellationTokenSource.Token);
}
private void OnClientConnected(object sender, MqttClientConnectedEventArgs eventArgs)
{
_logger.Info<MqttServer>("Client '{0}': Connected.", eventArgs.Client.ClientId);
ClientConnected?.Invoke(this, eventArgs);
}
private void OnClientDisconnected(object sender, MqttClientDisconnectedEventArgs eventArgs)
{
_logger.Info<MqttServer>("Client '{0}': Disconnected.", eventArgs.Client.ClientId);
ClientDisconnected?.Invoke(this, eventArgs);
}
private void OnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
ApplicationMessageReceived?.Invoke(this, e);
eventArgs.SessionTask = Task.Run(
async () => await _clientSessionsManager.RunClientSessionAsync(eventArgs.Client, _cancellationTokenSource.Token).ConfigureAwait(false),
_cancellationTokenSource.Token);
}
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttSubscriptionInterceptorContext.cs
+5 -3
View File
@@ -1,17 +1,19 @@
namespace MQTTnet.Server
using System;
namespace MQTTnet.Server
{
public class MqttSubscriptionInterceptorContext
{
public MqttSubscriptionInterceptorContext(string clientId, TopicFilter topicFilter)
{
ClientId = clientId;
TopicFilter = topicFilter;
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
}
public string ClientId { get; }
public TopicFilter TopicFilter { get; }
public bool AcceptSubscription { get; set; } = true;
public bool CloseConnection { get; set; }
Frameworks/MQTTnet.NetStandard/TopicFilter.cs
+1 -1
View File
@@ -4,7 +4,7 @@ namespace MQTTnet
{
public sealed class TopicFilter
{
public TopicFilter(string topic, MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce)
public TopicFilter(string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
{
Topic = topic;
QualityOfServiceLevel = qualityOfServiceLevel;