MqttBrokerNode.cs

using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading.Tasks;
using Waher.Networking;
using Waher.Networking.MQTT;
using Waher.Networking.Sniffers;
using Waher.Persistence;
using Waher.Persistence.Attributes;
using Waher.Runtime.Language;
using Waher.Things.Attributes;
using Waher.Things.DisplayableParameters;
using Waher.Things.Ip;
using Waher.Things.Mqtt.Model;
 
namespace Waher.Things.Mqtt
{
    public class MqttBrokerNode : IpHostPort, ICommunicationLayer, IEncryptedProperties
    {
        private MqttQualityOfService willQoS = MqttQualityOfService.AtLeastOnce;
        private string userName = string.Empty;
        private string password = string.Empty;
        private string willTopic = string.Empty;
        private string willData = string.Empty;
        private string brokerKey = null;
        private string connectionSubscription = "#";
        private bool willRetain = false;
        private bool trustServer = false;
 
        public MqttBrokerNode()
            : base()
        {
            this.Port = 8883;
            this.Tls = true;
        }
 
        [Page(1, "IP")]
        [Header(44, "Trust Server", 80)]
        [ToolTip(45, "If the remote server certificate should be trusted even if it is not valid.")]
        public bool TrustServer
        {
            get => this.trustServer;
            set => this.trustServer = value;
        }
 
        [Page(2, "MQTT")]
        [Header(3, "User Name:")]
        [ToolTip(4, "User name used during authentication process.")]
        [DefaultValueStringEmpty]
        public string UserName
        {
            get => this.userName;
            set => this.userName = value;
        }
 
        [Page(2, "MQTT")]
        [Header(5, "Password:")]
        [ToolTip(6, "Password used during authentication process. NOTE: Will be sent in clear text. Don't reuse passwords.")]
        [Masked]
        [Encrypted(32)]
        public string Password
        {
            get => this.password;
            set => this.password = value;
        }
 
        public string[] EncryptedProperties => new string[] { nameof(this.Password) };
 
        [Page(2, "MQTT")]
        [Header(46, "Connection Subscription:")]
        [ToolTip(47, "Subscription topic executed when connecting. Empty means no subscription will be performed. Multiple subjects can be comma-separated.")]
        [DefaultValue("#")]
        public string ConnectionSubscription
        {
            get => this.connectionSubscription;
            set => this.connectionSubscription = value;
        }
 
        [Page(10, "Last Will and Testament")]
        [Header(11, "Will Topic:")]
        [ToolTip(12, "When the connection is lost, a Last Will and Testament can be published on this topic to alert subscribers you've lost connection.")]
        [DefaultValueStringEmpty]
        public string WillTopic
        {
            get => this.willTopic;
            set => this.willTopic = value;
        }
 
        [Page(10, "Last Will and Testament")]
        [Header(13, "Will Data:")]
        [ToolTip(14, "When the connection is lost, this content will be published on the topic defined above.")]
        [DefaultValueStringEmpty]
        public string WillData
        {
            get => this.willData;
            set => this.willData = value;
        }
 
        [Page(10, "Last Will and Testament")]
        [Header(15, "Retain Will on topic.")]
        [ToolTip(16, "If the content published on the will should be retained on the topic.")]
        [DefaultValue(false)]
        public bool WillRetain
        {
            get => this.willRetain;
            set => this.willRetain = value;
        }
 
        [Page(10, "Last Will and Testament")]
        [Header(17, "Quality of Service:")]
        [ToolTip(18, "The quality of service used when sending the last will and testament.")]
        [DefaultValue(MqttQualityOfService.AtLeastOnce)]
        [Option(MqttQualityOfService.AtMostOnce, 19, "At most once")]
        [Option(MqttQualityOfService.AtLeastOnce, 20, "At least once")]
        [Option(MqttQualityOfService.ExactlyOnce, 19, "Exactly once")]
        public MqttQualityOfService WillQoS
        {
            get => this.willQoS;
            set => this.willQoS = value;
        }
 
        public override Task<string> GetTypeNameAsync(Language Language)
        {
            return Language.GetStringAsync(typeof(MqttBrokerNode), 1, "MQTT Broker");
        }
 
        public override Task<bool> AcceptsChildAsync(INode Child)
        {
            return Task.FromResult(Child is MqttTopicNode);
        }
 
        public override Task DestroyAsync()
        {
            if (!string.IsNullOrEmpty(this.brokerKey))
                MqttBrokers.DestroyBroker(this.brokerKey);
 
            return base.DestroyAsync();
        }
 
        [IgnoreMember]
        public string Key
        {
            get
            {
                string PrevKey = this.brokerKey;
                this.brokerKey = MqttBrokers.GetKey(this.Host, this.Port, this.Tls, this.trustServer, this.userName, this.password,
                    this.connectionSubscription);
 
                if (PrevKey != this.brokerKey && !string.IsNullOrEmpty(PrevKey))
                    MqttBrokers.DestroyBroker(PrevKey);
 
                return this.brokerKey;
            }
        }
 
        protected override Task NodeUpdated()
        {
            this.GetBroker();
 
            return base.NodeUpdated();
        }
 
        public Task<MqttBroker> GetBroker()
        {
            return MqttBrokers.GetBroker(this, this.Key, this.Host, this.Port, this.Tls, this.TrustServer, this.userName, this.password,
                this.connectionSubscription, this.willTopic, this.willData, this.willRetain, this.willQoS);
        }
 
        public MqttBroker GetCachedBroker()
        {
            return MqttBrokers.GetCachedBroker(this.Key);
        }
 
        public override async Task<bool> RemoveAsync(INode Child)
        {
            if (Child is MqttTopicNode Topic)
                (await this.GetBroker()).Remove(Topic.LocalTopic);
 
            return await base.RemoveAsync(Child);
        }
 
        #region ICommunicationLayer
 
        public bool DecoupledEvents => true;
 
        public void Add(ISniffer Sniffer)
        {
            this.GetBroker().Result.Client?.Add(Sniffer);   // TODO: Avoid blocking call
        }
 
        public void AddRange(IEnumerable<ISniffer> Sniffers)
        {
            this.GetBroker().Result.Client?.AddRange(Sniffers); // TODO: Avoid blocking call
        }
 
        public bool Remove(ISniffer Sniffer)
        {
            return this.GetBroker().Result.Client?.Remove(Sniffer) ?? false;    // TODO: Avoid blocking call
        }
 
        public ISniffer[] Sniffers
        {
            get { return this.GetBroker().Result.Client?.Sniffers ?? Array.Empty<ISniffer>(); } // TODO: Avoid blocking call
        }
 
        public bool HasSniffers
        {
            get { return this.GetBroker().Result.Client?.HasSniffers ?? false; }    // TODO: Avoid blocking call
        }
 
        public IEnumerator<ISniffer> GetEnumerator()
        {
            return new SnifferEnumerator(this.Sniffers);
        }
 
        IEnumerator IEnumerable.GetEnumerator()
        {
            return this.GetBroker().Result.Client?.GetEnumerator() ?? Array.Empty<ISniffer>().GetEnumerator();  // TODO: Avoid blocking call
        }
 
        public void ReceiveBinary(int Count)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.ReceiveBinary(Count);
        }
 
        public void ReceiveBinary(bool ConstantBuffer, byte[] Data)
        {
            this.ReceiveBinary(ConstantBuffer, Data, 0, Data.Length);
        }
 
        public void ReceiveBinary(bool ConstantBuffer, byte[] Data, int Offset, int Count)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.ReceiveBinary(ConstantBuffer, Data, Offset, Count);
        }
 
        public void TransmitBinary(int Count)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.TransmitBinary(Count);
        }
 
        public void TransmitBinary(bool ConstantBuffer, byte[] Data)
        {
            this.TransmitBinary(ConstantBuffer, Data, 0, Data.Length);
        }
 
        public void TransmitBinary(bool ConstantBuffer, byte[] Data, int Offset, int Count)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.TransmitBinary(ConstantBuffer, Data, Offset, Count);
        }
 
        public void ReceiveText(string Text)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.ReceiveText(Text);
        }
 
        public void TransmitText(string Text)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.TransmitText(Text);
        }
 
        public void Information(string Comment)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.Information(Comment);
        }
 
        public void Warning(string Warning)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.Warning(Warning);
        }
 
        public void Error(string Error)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.Error(Error);
        }
 
        public void Exception(Exception Exception)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.Exception(Exception);
        }
 
        public void Exception(string Exception)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.Exception(Exception);
        }
 
        public void ReceiveBinary(DateTime Timestamp, int Count)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.ReceiveBinary(Timestamp, Count);
        }
 
        public void ReceiveBinary(DateTime Timestamp, bool ConstantBuffer, byte[] Data)
        {
            this.ReceiveBinary(Timestamp, ConstantBuffer, Data, 0, Data.Length);
        }
 
        public void ReceiveBinary(DateTime Timestamp, bool ConstantBuffer, byte[] Data, int Offset, int Count)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.ReceiveBinary(Timestamp, ConstantBuffer, Data, Offset, Count);
        }
 
        public void TransmitBinary(DateTime Timestamp, int Count)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.TransmitBinary(Timestamp, Count);
        }
 
        public void TransmitBinary(DateTime Timestamp, bool ConstantBuffer, byte[] Data)
        {
            this.TransmitBinary(Timestamp, ConstantBuffer, Data, 0, Data.Length);
        }
 
        public void TransmitBinary(DateTime Timestamp, bool ConstantBuffer, byte[] Data, int Offset, int Count)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.TransmitBinary(Timestamp, ConstantBuffer, Data, Offset, Count);
        }
 
        public void ReceiveText(DateTime Timestamp, string Text)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.ReceiveText(Timestamp, Text);
        }
 
        public void TransmitText(DateTime Timestamp, string Text)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.TransmitText(Timestamp, Text);
        }
 
        public void Information(DateTime Timestamp, string Comment)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.Information(Timestamp, Comment);
        }
 
        public void Warning(DateTime Timestamp, string Warning)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.Warning(Timestamp, Warning);
        }
 
        public void Error(DateTime Timestamp, string Error)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.Error(Timestamp, Error);
        }
 
        public void Exception(DateTime Timestamp, string Exception)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.Exception(Timestamp, Exception);
        }
 
        public void Exception(DateTime Timestamp, Exception Exception)
        {
            MqttBroker Broker = this.GetCachedBroker();
            MqttClient Client = Broker?.Client;
            Client?.Exception(Timestamp, Exception);
        }
 
        #endregion
 
        public override Task<IEnumerable<ICommand>> Commands => this.GetCommands();
 
        public async Task<IEnumerable<ICommand>> GetCommands()
        {
            List<ICommand> Result = new List<ICommand>();
 
            Result.AddRange(await base.Commands);
            Result.Add(new ReconnectCommand((await this.GetBroker()).Client));
 
            return Result;
        }
 
        public async override Task<IEnumerable<Parameter>> GetDisplayableParametersAsync(Language Language, RequestOrigin Caller)
        {
            LinkedList<Parameter> Result = await base.GetDisplayableParametersAsync(Language, Caller) as LinkedList<Parameter>;
            MqttBroker Broker = await this.GetBroker();
 
            Result.AddLast(new StringParameter("State", await Language.GetStringAsync(typeof(MqttBrokerNode), 30, "State"),
                Broker.Client.State.ToString() ?? string.Empty));
 
            return Result;
        }
 
    }
}