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

Refactor logging and add benchmark.

Browse Source
This commit is contained in:
Christian
2018-05-11 22:00:54 +02:00
parent e14ea9a240
commit 8df8cb1eb6
27 changed files with 318 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Frameworks/MQTTnet.NetStandard/Server/MqttClientKeepAliveMonitor.cs
+8 -6
View File
@@ -14,15 +14,17 @@ namespace MQTTnet.Server
private readonly string _clientId;
private readonly Action _timeoutCallback;
private readonly IMqttNetLogger _logger;
private readonly IMqttNetChildLogger _logger;
private Task _workerTask;
public MqttClientKeepAliveMonitor(string clientId, Action timeoutCallback, IMqttNetLogger logger)
public MqttClientKeepAliveMonitor(string clientId, Action timeoutCallback, IMqttNetChildLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
_clientId = clientId;
_timeoutCallback = timeoutCallback;
_logger = logger;
_logger = logger.CreateChildLogger(nameof(MqttClientKeepAliveMonitor));
}
public TimeSpan LastPacketReceived => _lastPacketReceivedTracker.Elapsed;
@@ -59,7 +61,7 @@ namespace MQTTnet.Server
// Values described here: [MQTT-3.1.2-24].
if (_lastPacketReceivedTracker.Elapsed.TotalSeconds > keepAlivePeriod * 1.5D)
{
_logger.Warning(this, null, "Client '{0}': Did not receive any packet or keep alive signal.", _clientId);
_logger.Warning(null, "Client '{0}': Did not receive any packet or keep alive signal.", _clientId);
_timeoutCallback?.Invoke();
@@ -74,11 +76,11 @@ namespace MQTTnet.Server
}
catch (Exception exception)
{
_logger.Error(this, exception, "Client '{0}': Unhandled exception while checking keep alive timeouts.", _clientId);
_logger.Error(exception, "Client '{0}': Unhandled exception while checking keep alive timeouts.", _clientId);
}
finally
{
_logger.Verbose(this, "Client {0}: Stopped checking keep alive timeout.", _clientId);
_logger.Verbose("Client {0}: Stopped checking keep alive timeout.", _clientId);
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs
+11 -10
View File
@@ -17,15 +17,17 @@ namespace MQTTnet.Server
private readonly AsyncAutoResetEvent _queueAutoResetEvent = new AsyncAutoResetEvent();
private readonly IMqttServerOptions _options;
private readonly MqttClientSession _clientSession;
private readonly IMqttNetLogger _logger;
private readonly IMqttNetChildLogger _logger;
private Task _workerTask;
public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetLogger logger)
public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetChildLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
_options = options ?? throw new ArgumentNullException(nameof(options));
_clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttClientPendingMessagesQueue));
}
public int Count => _queue.Count;
@@ -70,7 +72,7 @@ namespace MQTTnet.Server
_queue.Enqueue(packet);
_queueAutoResetEvent.Set();
_logger.Verbose(this, "Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
_logger.Verbose("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
}
private async Task SendQueuedPacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
@@ -87,7 +89,7 @@ namespace MQTTnet.Server
}
catch (Exception exception)
{
_logger.Error(this, exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _clientSession.ClientId);
_logger.Error(exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _clientSession.ClientId);
}
}
@@ -113,25 +115,24 @@ namespace MQTTnet.Server
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { packet }, cancellationToken).ConfigureAwait(false);
_logger.Verbose<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).",
_clientSession.ClientId);
_logger.Verbose("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
}
catch (Exception exception)
{
if (exception is MqttCommunicationTimedOutException)
{
_logger.Warning(this, exception, "Sending publish packet failed: Timeout (ClientId: {0}).", _clientSession.ClientId);
_logger.Warning(exception, "Sending publish packet failed: Timeout (ClientId: {0}).", _clientSession.ClientId);
}
else if (exception is MqttCommunicationException)
{
_logger.Warning(this, exception, "Sending publish packet failed: Communication exception (ClientId: {0}).", _clientSession.ClientId);
_logger.Warning(exception, "Sending publish packet failed: Communication exception (ClientId: {0}).", _clientSession.ClientId);
}
else if (exception is OperationCanceledException)
{
}
else
{
_logger.Error(this, exception, "Sending publish packet failed (ClientId: {0}).", _clientSession.ClientId);
_logger.Error(exception, "Sending publish packet failed (ClientId: {0}).", _clientSession.ClientId);
}
if (packet is MqttPublishPacket publishPacket)
Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs
+14 -13
View File
@@ -17,7 +17,7 @@ namespace MQTTnet.Server
{
private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
private readonly MqttRetainedMessagesManager _retainedMessagesManager;
private readonly IMqttNetLogger _logger;
private readonly IMqttNetChildLogger _logger;
private readonly IMqttServerOptions _options;
private readonly MqttClientSessionsManager _sessionsManager;
@@ -30,15 +30,18 @@ namespace MQTTnet.Server
IMqttServerOptions options,
MqttClientSessionsManager sessionsManager,
MqttRetainedMessagesManager retainedMessagesManager,
IMqttNetLogger logger)
IMqttNetChildLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
_options = options ?? throw new ArgumentNullException(nameof(options));
_sessionsManager = sessionsManager;
_retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
ClientId = clientId;
_logger = logger.CreateChildLogger(nameof(MqttClientSession));
KeepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, StopDueToKeepAliveTimeout, _logger);
SubscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, sessionsManager.Server);
PendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
@@ -81,13 +84,11 @@ namespace MQTTnet.Server
}
catch (MqttCommunicationException exception)
{
_logger.Warning<MqttClientSession>(exception,
"Client '{0}': Communication exception while processing client packets.", ClientId);
_logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
}
catch (Exception exception)
{
_logger.Error<MqttClientSession>(exception,
"Client '{0}': Unhandled exception while processing client packets.", ClientId);
_logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
}
finally
{
@@ -126,7 +127,7 @@ namespace MQTTnet.Server
}
finally
{
_logger.Info<MqttClientSession>("Client '{0}': Session stopped.", ClientId);
_logger.Info("Client '{0}': Session stopped.", ClientId);
}
}
@@ -185,7 +186,7 @@ namespace MQTTnet.Server
private void StopDueToKeepAliveTimeout()
{
_logger.Info<MqttClientSession>("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId);
_logger.Info("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId);
Stop(MqttClientDisconnectType.NotClean);
}
@@ -207,11 +208,11 @@ namespace MQTTnet.Server
{
if (exception is MqttCommunicationException)
{
_logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
_logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
}
else
{
_logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
_logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
}
Stop(MqttClientDisconnectType.NotClean);
@@ -273,7 +274,7 @@ namespace MQTTnet.Server
return Task.FromResult(0);
}
_logger.Warning(this, null, "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
_logger.Warning(null, "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
Stop(MqttClientDisconnectType.NotClean);
return Task.FromResult(0);
Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs
+26 -22
View File
@@ -6,6 +6,7 @@ using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -14,15 +15,18 @@ namespace MQTTnet.Server
public sealed class MqttClientSessionsManager : IDisposable
{
private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
private readonly SemaphoreSlim _sessionsLock = new SemaphoreSlim(1, 1);
private readonly AsyncLock _sessionsLock = new AsyncLock();
private readonly MqttRetainedMessagesManager _retainedMessagesManager;
private readonly IMqttServerOptions _options;
private readonly IMqttNetLogger _logger;
private readonly IMqttNetChildLogger _logger;
public MqttClientSessionsManager(IMqttServerOptions options, MqttServer server, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
public MqttClientSessionsManager(IMqttServerOptions options, MqttServer server, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetChildLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttClientSessionsManager));
_options = options ?? throw new ArgumentNullException(nameof(options));
Server = server ?? throw new ArgumentNullException(nameof(server));
_retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
@@ -89,7 +93,7 @@ namespace MQTTnet.Server
}
catch (Exception exception)
{
_logger.Error<MqttClientSessionsManager>(exception, exception.Message);
_logger.Error(exception, exception.Message);
}
finally
{
@@ -100,7 +104,7 @@ namespace MQTTnet.Server
}
catch (Exception exception)
{
_logger.Error<MqttClientSessionsManager>(exception, exception.Message);
_logger.Error(exception, exception.Message);
}
Server.OnClientDisconnected(new ConnectedMqttClient
@@ -115,7 +119,7 @@ namespace MQTTnet.Server
public async Task StopAsync()
{
await _sessionsLock.WaitAsync().ConfigureAwait(false);
await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false);
try
{
foreach (var session in _sessions)
@@ -127,13 +131,13 @@ namespace MQTTnet.Server
}
finally
{
_sessionsLock.Release();
_sessionsLock.Exit();
}
}
public async Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
{
await _sessionsLock.WaitAsync().ConfigureAwait(false);
await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false);
try
{
return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
@@ -147,7 +151,7 @@ namespace MQTTnet.Server
}
finally
{
_sessionsLock.Release();
_sessionsLock.Exit();
}
}
@@ -161,7 +165,7 @@ namespace MQTTnet.Server
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
await _sessionsLock.WaitAsync().ConfigureAwait(false);
await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false);
try
{
if (!_sessions.TryGetValue(clientId, out var session))
@@ -173,7 +177,7 @@ namespace MQTTnet.Server
}
finally
{
_sessionsLock.Release();
_sessionsLock.Exit();
}
}
@@ -182,7 +186,7 @@ namespace MQTTnet.Server
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
await _sessionsLock.WaitAsync().ConfigureAwait(false);
await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false);
try
{
if (!_sessions.TryGetValue(clientId, out var session))
@@ -194,7 +198,7 @@ namespace MQTTnet.Server
}
finally
{
_sessionsLock.Release();
_sessionsLock.Exit();
}
}
@@ -222,7 +226,7 @@ namespace MQTTnet.Server
private async Task<GetOrCreateClientSessionResult> PrepareClientSessionAsync(MqttConnectPacket connectPacket)
{
await _sessionsLock.WaitAsync().ConfigureAwait(false);
await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false);
try
{
var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession);
@@ -236,11 +240,11 @@ namespace MQTTnet.Server
clientSession.Dispose();
clientSession = null;
_logger.Verbose<MqttClientSessionsManager>("Stopped existing session of client '{0}'.", connectPacket.ClientId);
_logger.Verbose("Stopped existing session of client '{0}'.", connectPacket.ClientId);
}
else
{
_logger.Verbose<MqttClientSessionsManager>("Reusing existing session of client '{0}'.", connectPacket.ClientId);
_logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
}
}
@@ -252,14 +256,14 @@ namespace MQTTnet.Server
clientSession = new MqttClientSession(connectPacket.ClientId, _options, this, _retainedMessagesManager, _logger);
_sessions[connectPacket.ClientId] = clientSession;
_logger.Verbose<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
_logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
}
return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
}
finally
{
_sessionsLock.Release();
_sessionsLock.Exit();
}
}
@@ -287,10 +291,10 @@ namespace MQTTnet.Server
}
catch (Exception exception)
{
_logger.Error<MqttClientSessionsManager>(exception, "Error while processing application message");
_logger.Error(exception, "Error while processing application message");
}
await _sessionsLock.WaitAsync().ConfigureAwait(false);
await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false);
try
{
foreach (var clientSession in _sessions.Values)
@@ -300,7 +304,7 @@ namespace MQTTnet.Server
}
finally
{
_sessionsLock.Release();
_sessionsLock.Exit();
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs
+29 -27
View File
@@ -4,19 +4,21 @@ using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Internal;
namespace MQTTnet.Server
{
public sealed class MqttRetainedMessagesManager : IDisposable
{
private readonly Dictionary<string, MqttApplicationMessage> _retainedMessages = new Dictionary<string, MqttApplicationMessage>();
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly IMqttNetLogger _logger;
private readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>();
private readonly AsyncLock _messagesLock = new AsyncLock();
private readonly IMqttNetChildLogger _logger;
private readonly IMqttServerOptions _options;
public MqttRetainedMessagesManager(IMqttServerOptions options, IMqttNetLogger logger)
public MqttRetainedMessagesManager(IMqttServerOptions options, IMqttNetChildLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttRetainedMessagesManager));
_options = options ?? throw new ArgumentNullException(nameof(options));
}
@@ -27,24 +29,24 @@ namespace MQTTnet.Server
return;
}
await _semaphore.WaitAsync().ConfigureAwait(false);
await _messagesLock.EnterAsync(CancellationToken.None).ConfigureAwait(false);
try
{
var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync();
var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false);
_retainedMessages.Clear();
_messages.Clear();
foreach (var retainedMessage in retainedMessages)
{
_retainedMessages[retainedMessage.Topic] = retainedMessage;
_messages[retainedMessage.Topic] = retainedMessage;
}
}
catch (Exception exception)
{
_logger.Error<MqttRetainedMessagesManager>(exception, "Unhandled exception while loading retained messages.");
_logger.Error(exception, "Unhandled exception while loading retained messages.");
}
finally
{
_semaphore.Release();
_messagesLock.Exit();
}
}
@@ -52,18 +54,18 @@ namespace MQTTnet.Server
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
await _semaphore.WaitAsync().ConfigureAwait(false);
await _messagesLock.EnterAsync(CancellationToken.None).ConfigureAwait(false);
try
{
await HandleMessageInternalAsync(clientId, applicationMessage);
}
catch (Exception exception)
{
_logger.Error<MqttRetainedMessagesManager>(exception, "Unhandled exception while handling retained messages.");
_logger.Error(exception, "Unhandled exception while handling retained messages.");
}
finally
{
_semaphore.Release();
_messagesLock.Exit();
}
}
@@ -71,10 +73,10 @@ namespace MQTTnet.Server
{
var retainedMessages = new List<MqttApplicationMessage>();
await _semaphore.WaitAsync().ConfigureAwait(false);
await _messagesLock.EnterAsync(CancellationToken.None).ConfigureAwait(false);
try
{
foreach (var retainedMessage in _retainedMessages.Values)
foreach (var retainedMessage in _messages.Values)
{
foreach (var topicFilter in topicFilters)
{
@@ -90,7 +92,7 @@ namespace MQTTnet.Server
}
finally
{
_semaphore.Release();
_messagesLock.Exit();
}
return retainedMessages;
@@ -98,7 +100,7 @@ namespace MQTTnet.Server
public void Dispose()
{
_semaphore?.Dispose();
_messagesLock?.Dispose();
}
private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage)
@@ -107,37 +109,37 @@ namespace MQTTnet.Server
if (applicationMessage.Payload?.Any() == false)
{
saveIsRequired = _retainedMessages.Remove(applicationMessage.Topic);
_logger.Info<MqttRetainedMessagesManager>("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
saveIsRequired = _messages.Remove(applicationMessage.Topic);
_logger.Info("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
else
{
if (!_retainedMessages.ContainsKey(applicationMessage.Topic))
if (!_messages.ContainsKey(applicationMessage.Topic))
{
_retainedMessages[applicationMessage.Topic] = applicationMessage;
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
else
{
var existingMessage = _retainedMessages[applicationMessage.Topic];
var existingMessage = _messages[applicationMessage.Topic];
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
{
_retainedMessages[applicationMessage.Topic] = applicationMessage;
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
}
_logger.Info<MqttRetainedMessagesManager>("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
_logger.Info("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
if (!saveIsRequired)
{
_logger.Verbose<MqttRetainedMessagesManager>("Skipped saving retained messages because no changes were detected.");
_logger.Verbose("Skipped saving retained messages because no changes were detected.");
}
if (saveIsRequired && _options.Storage != null)
{
await _options.Storage.SaveRetainedMessagesAsync(_retainedMessages.Values.ToList());
await _options.Storage.SaveRetainedMessagesAsync(_messages.Values.ToList()).ConfigureAwait(false);
}
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs
+8 -7
View File
@@ -11,16 +11,17 @@ namespace MQTTnet.Server
public class MqttServer : IMqttServer
{
private readonly ICollection<IMqttServerAdapter> _adapters;
private readonly IMqttNetLogger _logger;
private readonly IMqttNetChildLogger _logger;
private MqttClientSessionsManager _clientSessionsManager;
private MqttRetainedMessagesManager _retainedMessagesManager;
private CancellationTokenSource _cancellationTokenSource;
public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetChildLogger logger)
{
if (adapters == null) throw new ArgumentNullException(nameof(adapters));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttServer));
_adapters = adapters.ToList();
}
@@ -90,7 +91,7 @@ namespace MQTTnet.Server
await adapter.StartAsync(Options);
}
_logger.Info<MqttServer>("Started.");
_logger.Info("Started.");
Started?.Invoke(this, new MqttServerStartedEventArgs());
}
@@ -114,7 +115,7 @@ namespace MQTTnet.Server
await _clientSessionsManager.StopAsync();
_logger.Info<MqttServer>("Stopped.");
_logger.Info("Stopped.");
}
finally
{
@@ -129,13 +130,13 @@ namespace MQTTnet.Server
internal void OnClientConnected(ConnectedMqttClient client)
{
_logger.Info<MqttServer>("Client '{0}': Connected.", client.ClientId);
_logger.Info("Client '{0}': Connected.", client.ClientId);
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(client));
}
internal void OnClientDisconnected(ConnectedMqttClient client, bool wasCleanDisconnect)
{
_logger.Info<MqttServer>("Client '{0}': Disconnected (clean={1}).", client.ClientId, wasCleanDisconnect);
_logger.Info("Client '{0}': Disconnected (clean={1}).", client.ClientId, wasCleanDisconnect);
ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client, wasCleanDisconnect));
}