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 storage of creation time.

Browse Source
This commit is contained in:
Christian
2022-08-13 09:28:16 +02:00
parent 9152fe637e
commit ebbf50b41f
10 changed files with 231 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Samples/Server/Server_Retained_Messages_Samples.cs
+2 -1
View File
@@ -7,6 +7,7 @@
// ReSharper disable InconsistentNaming
using System.Text.Json;
using MQTTnet.Server;
namespace MQTTnet.Samples.Server;
@@ -32,7 +33,7 @@ public static class Server_Retained_Messages_Samples
{
try
{
eventArgs.LoadedRetainedMessages = await JsonSerializer.DeserializeAsync<List<MqttApplicationMessage>>(File.OpenRead(storePath));
eventArgs.LoadedRetainedMessages = await JsonSerializer.DeserializeAsync<List<MqttRetainedMessage>>(File.OpenRead(storePath));
Console.WriteLine("Retained messages loaded.");
}
catch (FileNotFoundException)
Source/MQTTnet.TestApp/ServerTest.cs
+3 -5
View File
@@ -4,12 +4,10 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Implementations;
using MQTTnet.Protocol;
using MQTTnet.Server;
@@ -80,15 +78,15 @@ namespace MQTTnet.TestApp
mqttServer.LoadingRetainedMessageAsync += e =>
{
List<MqttApplicationMessage> retainedMessages;
List<MqttRetainedMessage> retainedMessages;
if (File.Exists(Filename))
{
var json = File.ReadAllText(Filename);
retainedMessages = JsonConvert.DeserializeObject<List<MqttApplicationMessage>>(json);
retainedMessages = JsonConvert.DeserializeObject<List<MqttRetainedMessage>>(json);
}
else
{
retainedMessages = new List<MqttApplicationMessage>();
retainedMessages = new List<MqttRetainedMessage>();
}
e.LoadedRetainedMessages = retainedMessages;
Source/MQTTnet.Tests/Server/General.cs
+1 -1
View File
@@ -394,7 +394,7 @@ namespace MQTTnet.Tests.Server
{
using (var testEnvironment = CreateTestEnvironment())
{
List<MqttApplicationMessage> savedRetainedMessages = null;
ICollection<MqttRetainedMessage> savedRetainedMessages = null;
var s = await testEnvironment.StartServer();
s.RetainedMessageChangedAsync += e =>
Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs
+1 -1
View File
@@ -9,6 +9,6 @@ namespace MQTTnet.Server
{
public sealed class LoadingRetainedMessagesEventArgs : EventArgs
{
public List<MqttApplicationMessage> LoadedRetainedMessages { get; set; } = new List<MqttApplicationMessage>();
public ICollection<MqttRetainedMessage> LoadedRetainedMessages { get; set; } = new List<MqttRetainedMessage>();
}
}
Source/MQTTnet/Server/Events/RetainedMessageChangeType.cs
+15
View File
@@ -0,0 +1,15 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
namespace MQTTnet.Server
{
public enum RetainedMessageChangeType
{
Created,
Updated,
Deleted
}
}
Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs
+12 -5
View File
@@ -9,17 +9,24 @@ namespace MQTTnet.Server
{
public sealed class RetainedMessageChangedEventArgs : EventArgs
{
public RetainedMessageChangedEventArgs(string clientId, MqttApplicationMessage changedRetainedMessage, List<MqttApplicationMessage> storedRetainedMessages)
public RetainedMessageChangedEventArgs(
string clientId,
MqttRetainedMessage changedMessage,
RetainedMessageChangeType changeType,
List<MqttRetainedMessage> allMessages)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
ChangedRetainedMessage = changedRetainedMessage ?? throw new ArgumentNullException(nameof(changedRetainedMessage));
StoredRetainedMessages = storedRetainedMessages ?? throw new ArgumentNullException(nameof(storedRetainedMessages));
ChangedRetainedMessage = changedMessage ?? throw new ArgumentNullException(nameof(changedMessage));
ChangeType = changeType;
StoredRetainedMessages = allMessages ?? throw new ArgumentNullException(nameof(allMessages));
}
public MqttApplicationMessage ChangedRetainedMessage { get; }
public RetainedMessageChangeType ChangeType { get; }
public MqttRetainedMessage ChangedRetainedMessage { get; }
public string ClientId { get; }
public List<MqttApplicationMessage> StoredRetainedMessages { get; }
public List<MqttRetainedMessage> StoredRetainedMessages { get; }
}
}
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+2 -2
View File
@@ -58,7 +58,7 @@ namespace MQTTnet.Server
_eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
}
public async Task CloseAllConnectionsAsync()
public async Task Stop()
{
List<MqttClient> connections;
lock (_clients)
@@ -131,7 +131,7 @@ namespace MQTTnet.Server
{
if (applicationMessage.Retain)
{
await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false);
await _retainedMessagesManager.HandleMessage(senderId, applicationMessage).ConfigureAwait(false);
}
var deliveryCount = 0;
Source/MQTTnet/Server/Internal/MqttRetainedMessage.cs
+18
View File
@@ -0,0 +1,18 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
namespace MQTTnet.Server
{
public sealed class MqttRetainedMessage
{
public MqttApplicationMessage ApplicationMessage { get; set; }
/// <summary>
/// Gets the timestamp in UTC when this retained message eas published.
/// </summary>
public DateTime PublishedTimestamp { get; set; }
}
}
Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs
+173 -70
View File
@@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information.
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
@@ -15,47 +16,42 @@ namespace MQTTnet.Server
{
public sealed class MqttRetainedMessagesManager
{
readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>(4096);
readonly AsyncLock _storageAccessLock = new AsyncLock();
readonly MqttServerEventContainer _eventContainer;
readonly MqttNetSourceLogger _logger;
readonly ConcurrentDictionary<string, MqttRetainedMessage> _messages = new ConcurrentDictionary<string, MqttRetainedMessage>();
readonly AsyncLock _storageAccessLock = new AsyncLock();
CancellationTokenSource _cancellationToken;
public MqttRetainedMessagesManager(MqttServerEventContainer eventContainer, IMqttNetLogger logger)
{
_eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
if (logger == null) throw new ArgumentNullException(nameof(logger));
if (logger == null)
{
throw new ArgumentNullException(nameof(logger));
}
_logger = logger.WithSource(nameof(MqttRetainedMessagesManager));
}
public async Task Start()
{
try
{
var eventArgs = new LoadingRetainedMessagesEventArgs();
await _eventContainer.LoadingRetainedMessagesEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
lock (_messages)
{
_messages.Clear();
if (eventArgs.LoadedRetainedMessages != null)
{
foreach (var retainedMessage in eventArgs.LoadedRetainedMessages)
{
_messages[retainedMessage.Topic] = retainedMessage;
}
}
}
}
catch (Exception exception)
public async Task ClearMessages()
{
_messages.Clear();
using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
{
_logger.Error(exception, "Unhandled exception while loading retained messages.");
await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
}
}
public async Task UpdateMessage(string clientId, MqttApplicationMessage applicationMessage)
public Task<IList<MqttApplicationMessage>> GetMessages()
{
var result = new List<MqttApplicationMessage>(_messages.Select(m => m.Value.ApplicationMessage));
return Task.FromResult((IList<MqttApplicationMessage>)result);
}
public async Task HandleMessage(string clientId, MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
@@ -64,78 +60,185 @@ namespace MQTTnet.Server
try
{
List<MqttApplicationMessage> messagesForSave = null;
List<MqttRetainedMessage> allMessages = null;
var saveIsRequired = false;
lock (_messages)
var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
var changeType = RetainedMessageChangeType.Created;
var retainedMessage = new MqttRetainedMessage
{
var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
ApplicationMessage = applicationMessage,
PublishedTimestamp = DateTime.UtcNow
};
if (!hasPayload)
{
saveIsRequired = _messages.Remove(applicationMessage.Topic);
_logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
else
{
if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage))
MqttRetainedMessage changedMessage;
if (!hasPayload)
{
// Messages without a payload will remove the message.
changeType = RetainedMessageChangeType.Deleted;
saveIsRequired = _messages.TryRemove(applicationMessage.Topic, out changedMessage);
_logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
else
{
var messageChanged = false;
_messages.AddOrUpdate(
applicationMessage.Topic,
_ =>
{
_messages[applicationMessage.Topic] = applicationMessage;
changeType = RetainedMessageChangeType.Created;
changedMessage = retainedMessage;
saveIsRequired = true;
}
else
return retainedMessage;
},
(_, existingMessage) =>
{
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? PlatformAbstractionLayer.EmptyByteArray))
var hasSamePayload = existingMessage.ApplicationMessage.Payload.SequenceEqual(applicationMessage.Payload ?? PlatformAbstractionLayer.EmptyByteArray);
var hasSameQoSLevel = existingMessage.ApplicationMessage.QualityOfServiceLevel == applicationMessage.QualityOfServiceLevel;
if (!hasSameQoSLevel || !hasSamePayload)
{
_messages[applicationMessage.Topic] = applicationMessage;
changeType = RetainedMessageChangeType.Updated;
saveIsRequired = true;
messageChanged = true;
return retainedMessage;
}
}
_logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
// Keep the existing message.
return existingMessage;
});
if (saveIsRequired)
if (messageChanged)
{
messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
_logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
}
if (saveIsRequired)
{
using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
{
var eventArgs = new RetainedMessageChangedEventArgs(clientId, applicationMessage, messagesForSave);
await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
allMessages = new List<MqttRetainedMessage>(_messages.Values);
}
if (saveIsRequired)
{
await SaveMessages(clientId, retainedMessage, changeType, allMessages).ConfigureAwait(false);
}
}
catch (Exception exception)
{
_logger.Error(exception, "Unhandled exception while handling retained messages.");
}
}
public Task<IList<MqttApplicationMessage>> GetMessages()
{
lock (_messages)
{
var result = new List<MqttApplicationMessage>(_messages.Values);
return Task.FromResult((IList<MqttApplicationMessage>)result);
_logger.Error(exception, "Error while handling retained messages.");
}
}
public async Task ClearMessages()
public async Task Start()
{
lock (_messages)
_cancellationToken = new CancellationTokenSource();
await LoadMessages().ConfigureAwait(false);
_ = Task.Run(() => RemoveExpiredMessagesLoop(_cancellationToken.Token), _cancellationToken.Token);
}
public Task Stop()
{
_cancellationToken?.Cancel();
_cancellationToken?.Dispose();
_cancellationToken = null;
return PlatformAbstractionLayer.CompletedTask;
}
async Task LoadMessages()
{
if (!_eventContainer.LoadingRetainedMessagesEvent.HasHandlers)
{
_messages.Clear();
return;
}
var eventArgs = new LoadingRetainedMessagesEventArgs();
await _eventContainer.LoadingRetainedMessagesEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
_messages.Clear();
if (eventArgs.LoadedRetainedMessages != null)
{
foreach (var retainedMessage in eventArgs.LoadedRetainedMessages)
{
_messages[retainedMessage.ApplicationMessage.Topic] = retainedMessage;
}
}
}
async Task RemoveExpiredMessages()
{
var expiredMessages = new List<MqttRetainedMessage>();
var now = DateTime.UtcNow;
foreach (var retainedMessage in _messages)
{
if (retainedMessage.Value.ApplicationMessage.MessageExpiryInterval == 0)
{
// The feature is not used.
continue;
}
var messageAge = (int)(now - retainedMessage.Value.PublishedTimestamp).TotalSeconds;
if (messageAge < retainedMessage.Value.ApplicationMessage.MessageExpiryInterval)
{
// The message is not expired yet.
continue;
}
expiredMessages.Add(retainedMessage.Value);
}
foreach (var expiredMessage in expiredMessages)
{
if (_messages.TryRemove(expiredMessage.ApplicationMessage.Topic, out _))
{
_logger.Verbose("Server cleared retained message for topic '{0}' due to expiration.", expiredMessage.ApplicationMessage.Topic);
await SaveMessages(null, expiredMessage, RetainedMessageChangeType.Deleted, new List<MqttRetainedMessage>(_messages.Values));
}
}
}
async Task RemoveExpiredMessagesLoop(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
try
{
await RemoveExpiredMessages().ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.Error(exception, "Error while removing expired retained messages.");
}
finally
{
// The expiry interval is specified in seconds so that a message may gets expired
// every second.
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
}
}
}
async Task SaveMessages(string clientId, MqttRetainedMessage changedMessage, RetainedMessageChangeType changeType, List<MqttRetainedMessage> allMessages)
{
using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
{
await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
if (_eventContainer.RetainedMessageChangedEvent.HasHandlers)
{
var eventArgs = new RetainedMessageChangedEventArgs(clientId, changedMessage, changeType, allMessages);
await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
}
}
}
}
}
Source/MQTTnet/Server/MqttServer.cs
+4 -3
View File
@@ -287,8 +287,9 @@ namespace MQTTnet.Server
_cancellationTokenSource.Cancel(false);
await _clientSessionsManager.CloseAllConnectionsAsync().ConfigureAwait(false);
await _clientSessionsManager.Stop().ConfigureAwait(false);
await _retainedMessagesManager.Stop().ConfigureAwait(false);
foreach (var adapter in _adapters)
{
adapter.ClientHandler = null;
@@ -357,7 +358,7 @@ namespace MQTTnet.Server
ThrowIfDisposed();
ThrowIfNotStarted();
return _retainedMessagesManager?.UpdateMessage(null, retainedMessage);
return _retainedMessagesManager?.HandleMessage(null, retainedMessage);
}
protected override void Dispose(bool disposing)