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

Expose retained messages to server interface.

Browse Source
This commit is contained in:
Christian Kratky
2018-10-19 23:56:32 +02:00
parent 4c76a898b6
commit aad3ab903f
4 changed files with 75 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/IMqttServer.cs
+6 -2
View File
@@ -16,14 +16,18 @@ namespace MQTTnet.Server
IMqttServerOptions Options { get; }
[Obsolete("This method is no longer async. Use the not async method.")]
Task<IList<IMqttClientSessionStatus>> GetClientSessionsStatusAsync();
IList<IMqttClientSessionStatus> GetClientSessionsStatus();
IList<MqttApplicationMessage> GetRetainedMessages();
Task ClearRetainedMessagesAsync();
Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters);
Task UnsubscribeAsync(string clientId, IList<string> topicFilters);
Task StartAsync(IMqttServerOptions options);
Task StopAsync();
Task ClearRetainedMessagesAsync();
}
}
Source/MQTTnet/Server/MqttClientSessionsManager.cs
+2 -2
View File
@@ -65,7 +65,7 @@ namespace MQTTnet.Server
return Task.Run(() => RunSession(clientAdapter, _cancellationToken), _cancellationToken);
}
public Task<IList<IMqttClientSessionStatus>> GetClientStatusAsync()
public IList<IMqttClientSessionStatus> GetClientStatus()
{
var result = new List<IMqttClientSessionStatus>();
@@ -77,7 +77,7 @@ namespace MQTTnet.Server
result.Add(status);
}
return Task.FromResult((IList<IMqttClientSessionStatus>)result);
return result;
}
public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
+56 -47
View File
@@ -52,7 +52,51 @@ namespace MQTTnet.Server
try
{
await HandleMessageInternalAsync(clientId, applicationMessage).ConfigureAwait(false);
List<MqttApplicationMessage> messagesForSave = null;
lock (_messages)
{
var saveIsRequired = false;
if (applicationMessage.Payload?.Length == 0)
{
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))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
else
{
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
}
_logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
if (saveIsRequired)
{
messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
}
}
if (messagesForSave == null)
{
_logger.Verbose("Skipped saving retained messages because no changes were detected.");
return;
}
if (_options.Storage != null)
{
await _options.Storage.SaveRetainedMessagesAsync(messagesForSave).ConfigureAwait(false);
}
}
catch (Exception exception)
{
@@ -84,6 +128,14 @@ namespace MQTTnet.Server
return retainedMessages;
}
public IList<MqttApplicationMessage> GetMessages()
{
lock (_messages)
{
return _messages.Values.ToList();
}
}
public Task ClearMessagesAsync()
{
lock (_messages)
@@ -91,55 +143,12 @@ namespace MQTTnet.Server
_messages.Clear();
}
return _options.Storage.SaveRetainedMessagesAsync(new List<MqttApplicationMessage>());
}
private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage)
{
var saveIsRequired = false;
lock (_messages)
if (_options.Storage != null)
{
if (applicationMessage.Payload?.Length == 0)
{
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))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
else
{
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
}
_logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
return _options.Storage.SaveRetainedMessagesAsync(new List<MqttApplicationMessage>());
}
if (!saveIsRequired)
{
_logger.Verbose("Skipped saving retained messages because no changes were detected.");
}
if (saveIsRequired && _options.Storage != null)
{
List<MqttApplicationMessage> messages;
lock (_messages)
{
messages = _messages.Values.ToList();
}
await _options.Storage.SaveRetainedMessagesAsync(messages).ConfigureAwait(false);
}
return Task.FromResult((object)null);
}
}
}
Source/MQTTnet/Server/MqttServer.cs
+11 -1
View File
@@ -41,7 +41,17 @@ namespace MQTTnet.Server
public Task<IList<IMqttClientSessionStatus>> GetClientSessionsStatusAsync()
{
return _clientSessionsManager.GetClientStatusAsync();
return Task.FromResult(_clientSessionsManager.GetClientStatus());
}
public IList<IMqttClientSessionStatus> GetClientSessionsStatus()
{
return _clientSessionsManager.GetClientStatus();
}
public IList<MqttApplicationMessage> GetRetainedMessages()
{
return _retainedMessagesManager.GetMessages();
}
public Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)