MQTTnet/Samples/Server/Server_Retained_Messages_Samples.cs

// 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.

// ReSharper disable UnusedType.Global
// ReSharper disable UnusedMember.Global
// ReSharper disable InconsistentNaming

using System.Buffers;
using System.Text.Json;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;

namespace MQTTnet.Samples.Server;

public static class Server_Retained_Messages_Samples
{
    public static async Task Persist_Retained_Messages()
    {
        /*
         * This sample starts a simple MQTT server which will store all retained messages in a file.
         */

        var storePath = Path.Combine(Path.GetTempPath(), "RetainedMessages.json");

        var mqttServerFactory = new MqttServerFactory();

        // Due to security reasons the "default" endpoint (which is unencrypted) is not enabled by default!
        var mqttServerOptions = mqttServerFactory.CreateServerOptionsBuilder().WithDefaultEndpoint().Build();

        using var server = mqttServerFactory.CreateMqttServer(mqttServerOptions);
        // Make sure that the server will load the retained messages.
        server.LoadingRetainedMessageAsync += async eventArgs =>
        {
            try
            {
                var models = await JsonSerializer.DeserializeAsync<List<MqttRetainedMessageModel>>(File.OpenRead(storePath)) ?? new List<MqttRetainedMessageModel>();
                var retainedMessages = models.Select(m => m.ToApplicationMessage()).ToList();

                eventArgs.LoadedRetainedMessages = retainedMessages;
                Console.WriteLine("Retained messages loaded.");
            }
            catch (FileNotFoundException)
            {
                // Ignore because nothing is stored yet.
                Console.WriteLine("No retained messages stored yet.");
            }
            catch (Exception exception)
            {
                Console.WriteLine(exception);
            }
        };

        // Make sure to persist the changed retained messages.
        server.RetainedMessageChangedAsync += async eventArgs =>
        {
            try
            {
                // This sample uses the property _StoredRetainedMessages_ which will contain all(!) retained messages.
                // The event args also contain the affected retained message (property ChangedRetainedMessage). This can be
                // used to write all retained messages to dedicated files etc. Then all files must be loaded and a full list
                // of retained messages must be provided in the loaded event.

                var models = eventArgs.StoredRetainedMessages.Select(MqttRetainedMessageModel.Create);

                var buffer = JsonSerializer.SerializeToUtf8Bytes(models);
                await File.WriteAllBytesAsync(storePath, buffer);
                Console.WriteLine("Retained messages saved.");
            }
            catch (Exception exception)
            {
                Console.WriteLine(exception);
            }
        };

        // Make sure to clear the retained messages when they are all deleted via API.
        server.RetainedMessagesClearedAsync += _ =>
        {
            File.Delete(storePath);
            return Task.CompletedTask;
        };

        await server.StartAsync();

        Console.WriteLine("Press Enter to exit.");
        Console.ReadLine();
    }

    sealed class MqttRetainedMessageModel
    {
        public string? ContentType { get; set; }
        public byte[]? CorrelationData { get; set; }
        public byte[]? Payload { get; set; }
        public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; }
        public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
        public string? ResponseTopic { get; set; }
        public string? Topic { get; set; }
        public List<MqttUserProperty>? UserProperties { get; set; }

        public static MqttRetainedMessageModel Create(MqttApplicationMessage message)
        {
            ArgumentNullException.ThrowIfNull(message);

            return new MqttRetainedMessageModel
            {
                Topic = message.Topic,

                // Create a copy of the buffer from the payload segment because
                // it cannot be serialized and deserialized with the JSON serializer.
                Payload = message.Payload.ToArray(),
                UserProperties = message.UserProperties,
                ResponseTopic = message.ResponseTopic,
                CorrelationData = message.CorrelationData,
                ContentType = message.ContentType,
                PayloadFormatIndicator = message.PayloadFormatIndicator,
                QualityOfServiceLevel = message.QualityOfServiceLevel

                // Other properties like "Retain" are not if interest in the storage.
                // That's why a custom model makes sense.
            };
        }

        public MqttApplicationMessage ToApplicationMessage()
        {
            return new MqttApplicationMessage
            {
                Topic = Topic,
                PayloadSegment = new ArraySegment<byte>(Payload ?? Array.Empty<byte>()),
                PayloadFormatIndicator = PayloadFormatIndicator,
                ResponseTopic = ResponseTopic,
                CorrelationData = CorrelationData,
                ContentType = ContentType,
                UserProperties = UserProperties,
                QualityOfServiceLevel = QualityOfServiceLevel,
                Dup = false,
                Retain = true
            };
        }
    }
}