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

Add new method for reading a single retained message (#1676)

Browse Source
This commit is contained in:
Christian
2023-02-15 19:46:37 +01:00
committed by GitHub
parent 9cbd5dd778
commit b599a832da
3 changed files with 37 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs
+23 -8
View File
@@ -22,18 +22,19 @@ namespace MQTTnet.Server
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();
@@ -64,7 +65,7 @@ namespace MQTTnet.Server
{
List<MqttApplicationMessage> messagesForSave = null;
var saveIsRequired = false;
lock (_messages)
{
var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
@@ -83,7 +84,8 @@ namespace MQTTnet.Server
}
else
{
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? EmptyBuffer.Array))
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel ||
!existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? EmptyBuffer.Array))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
@@ -113,7 +115,7 @@ namespace MQTTnet.Server
_logger.Error(exception, "Unhandled exception while handling retained messages.");
}
}
public Task<IList<MqttApplicationMessage>> GetMessages()
{
lock (_messages)
@@ -123,6 +125,19 @@ namespace MQTTnet.Server
}
}
public Task<MqttApplicationMessage> GetMessage(string topic)
{
lock (_messages)
{
if (_messages.TryGetValue(topic, out var message))
{
return Task.FromResult(message);
}
return null;
}
}
public async Task ClearMessages()
{
lock (_messages)
@@ -136,4 +151,4 @@ namespace MQTTnet.Server
}
}
}
}
}
Source/MQTTnet/Server/MqttServer.cs
+13 -1
View File
@@ -204,6 +204,18 @@ namespace MQTTnet.Server
return _retainedMessagesManager.GetMessages();
}
public Task<MqttApplicationMessage> GetRetainedMessageAsync(string topic)
{
if (topic == null)
{
throw new ArgumentNullException(nameof(topic));
}
ThrowIfNotStarted();
return _retainedMessagesManager.GetMessage(topic);
}
public Task<IList<MqttSessionStatus>> GetSessionsAsync()
{
ThrowIfNotStarted();
@@ -233,7 +245,7 @@ namespace MQTTnet.Server
}
var sessionItems = injectedApplicationMessage.CustomSessionItems ?? ServerSessionItems;
return _clientSessionsManager.DispatchApplicationMessage(
injectedApplicationMessage.SenderClientId,
sessionItems,