From aad3ab903f87b429d080bbab0a145b7fa6c5105b Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 19 Oct 2018 23:56:32 +0200 Subject: [PATCH] Expose retained messages to server interface. --- Source/MQTTnet/Server/IMqttServer.cs | 8 +- .../Server/MqttClientSessionsManager.cs | 4 +- .../Server/MqttRetainedMessagesManager.cs | 103 ++++++++++-------- Source/MQTTnet/Server/MqttServer.cs | 12 +- 4 files changed, 75 insertions(+), 52 deletions(-) diff --git a/Source/MQTTnet/Server/IMqttServer.cs b/Source/MQTTnet/Server/IMqttServer.cs index 55ba710d..3a45973b 100644 --- a/Source/MQTTnet/Server/IMqttServer.cs +++ b/Source/MQTTnet/Server/IMqttServer.cs @@ -16,14 +16,18 @@ namespace MQTTnet.Server IMqttServerOptions Options { get; } + [Obsolete("This method is no longer async. Use the not async method.")] Task> GetClientSessionsStatusAsync(); + IList GetClientSessionsStatus(); + + IList GetRetainedMessages(); + Task ClearRetainedMessagesAsync(); + Task SubscribeAsync(string clientId, IList topicFilters); Task UnsubscribeAsync(string clientId, IList topicFilters); Task StartAsync(IMqttServerOptions options); Task StopAsync(); - - Task ClearRetainedMessagesAsync(); } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index d8f4f141..400045b4 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -65,7 +65,7 @@ namespace MQTTnet.Server return Task.Run(() => RunSession(clientAdapter, _cancellationToken), _cancellationToken); } - public Task> GetClientStatusAsync() + public IList GetClientStatus() { var result = new List(); @@ -77,7 +77,7 @@ namespace MQTTnet.Server result.Add(status); } - return Task.FromResult((IList)result); + return result; } public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket) diff --git a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs index c2c8eb0d..bf5de984 100644 --- a/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/MqttRetainedMessagesManager.cs @@ -52,7 +52,51 @@ namespace MQTTnet.Server try { - await HandleMessageInternalAsync(clientId, applicationMessage).ConfigureAwait(false); + List 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(_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 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()); - } - - 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()); } - if (!saveIsRequired) - { - _logger.Verbose("Skipped saving retained messages because no changes were detected."); - } - - if (saveIsRequired && _options.Storage != null) - { - List messages; - lock (_messages) - { - messages = _messages.Values.ToList(); - } - - await _options.Storage.SaveRetainedMessagesAsync(messages).ConfigureAwait(false); - } + return Task.FromResult((object)null); } } } diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 4086b18a..a960ab00 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -41,7 +41,17 @@ namespace MQTTnet.Server public Task> GetClientSessionsStatusAsync() { - return _clientSessionsManager.GetClientStatusAsync(); + return Task.FromResult(_clientSessionsManager.GetClientStatus()); + } + + public IList GetClientSessionsStatus() + { + return _clientSessionsManager.GetClientStatus(); + } + + public IList GetRetainedMessages() + { + return _retainedMessagesManager.GetMessages(); } public Task SubscribeAsync(string clientId, IList topicFilters)