MqttActorTcp.cs

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Xml;
using TAG.Simulator.ObjectModel.Actors;
using Waher.Content.Xml;
using Waher.Networking.Sniffers;
using Waher.Networking.MQTT;
using Waher.Persistence;
using Waher.Persistence.Filters;
using Waher.Script;
using Waher.Script.Persistence.SQL;
 
namespace TAG.Simulator.MQTT.Actors
{
    public class MqttActorTcp : Actor
    {
        public const string MqttNamespace = "http://lab.tagroot.io/Schema/ComSim/MQTT.xsd";
 
        public const string MqttSchema = "TAG.Simulator.MQTT.Schema.ComSimMqtt.xsd";
 
        private KeyValuePair<string, MqttQualityOfService>[] subscriptions;
        private AccountCredentials credentials;
        private MqttClient client;
        private ISniffer sniffer;
        private string domain;
        private string userName;
        private string password;
        private int port;
        private bool encrypted;
        private bool trustServer;
        private bool isOnline = false;
        private TaskCompletionSource<bool> connected;
 
        public MqttActorTcp(ISimulationNode Parent, Model Model)
            : base(Parent, Model)
        {
        }
 
        public MqttActorTcp(ISimulationNode Parent, Model Model, int InstanceIndex, string InstanceId)
            : base(Parent, Model, InstanceIndex, InstanceId)
        {
        }
 
        public override string LocalName => nameof(MqttActorTcp);
 
        public override string Namespace => MqttNamespace;
 
        public override string SchemaResource => MqttSchema;
 
        public bool IsOnline => this.isOnline;
 
        public bool TrustServer => this.trustServer;
 
        public string Domain => this.domain;
 
        public MqttClient Client => this.client;
 
        public override ISimulationNode Create(ISimulationNode Parent, Model Model)
        {
            return new MqttActorTcp(Parent, Model);
        }
 
        public override Task FromXml(XmlElement Definition)
        {
            this.domain = XML.Attribute(Definition, "domain");
            this.encrypted = XML.Attribute(Definition, "encrypted", false);
            this.port = XML.Attribute(Definition, "port", this.encrypted ? 8883 : 1883);
            this.userName = XML.Attribute(Definition, "userName");
            this.password = XML.Attribute(Definition, "password");
            this.trustServer = XML.Attribute(Definition, "trustServer", false);
 
            return base.FromXml(Definition);
        }
 
        public override async Task<Actor> CreateInstanceAsync(int InstanceIndex, string InstanceId)
        {
            MqttActorTcp Result = new MqttActorTcp(this, this.Model, InstanceIndex, InstanceId)
            {
                domain = this.domain,
                encrypted = this.encrypted,
                port = this.port,
                userName = this.userName + InstanceIndex.ToString(),
                password = this.password,
                trustServer = this.trustServer
            };
 
            Variables Variables = new Variables();
            ObjectProperties Properties = new ObjectProperties(Result, Variables);
            List<KeyValuePair<string, MqttQualityOfService>> Topics = new List<KeyValuePair<string, MqttQualityOfService>>();
 
            foreach (ISimulationNode Node in this.Children)
            {
                if (Node is Subscribe Subscribe)
                {
                    string Topic = await Expression.TransformAsync(Subscribe.Topic, "{", "}", Properties);
                    Topics.Add(new KeyValuePair<string, MqttQualityOfService>(Topic, Subscribe.QoS));
                }
            }
 
            Result.subscriptions = Topics.ToArray();
 
            return Result;
        }
 
        public override async Task InitializeInstance()
        {
            this.credentials = await Database.FindFirstIgnoreRest<AccountCredentials>(new FilterAnd(
                new FilterFieldEqualTo("Domain", this.domain),
                new FilterFieldEqualTo("UserName", this.userName)));
 
            if (this.credentials is null)
            {
                this.credentials = new AccountCredentials()
                {
                    Domain = this.domain,
                    UserName = this.userName,
                    Password = string.IsNullOrEmpty(this.password) ? string.Empty : await this.Model.GetKey(this.password, this.userName)
                };
            }
 
            this.sniffer = this.Model.GetSniffer(this.userName);
 
            if (this.sniffer is null)
                this.client = new MqttClient(this.domain, this.port, this.encrypted, this.userName, this.credentials.Password);
            else
                this.client = new MqttClient(this.domain, this.port, this.encrypted, this.userName, this.credentials.Password, this.sniffer);
 
            this.client.TrustServer = this.trustServer;
 
            this.client.OnStateChanged += this.Client_OnStateChanged;
            this.client.OnConnectionError += this.Client_OnConnectionError;
            this.client.OnError += this.Client_OnError;
            this.client.OnContentReceived += this.Client_OnContentReceived;
            this.client.OnPing += this.Client_OnPing;
            this.client.OnPingResponse += this.Client_OnPingResponse;
            this.client.OnPublished += this.Client_OnPublished;
            this.client.OnSubscribed += this.Client_OnSubscribed;
            this.client.OnUnsubscribed += this.Client_OnUnsubscribed;
 
            this.connected = new TaskCompletionSource<bool>();
        }
 
        private Task Client_OnUnsubscribed(object Sender, ushort PacketIdentifier)
        {
            this.Model.ExternalEvent(this, "OnUnsubscribed",
                new KeyValuePair<string, object>("PacketIdentifier", PacketIdentifier),
                new KeyValuePair<string, object>("Client", this.client));
 
            return Task.CompletedTask;
        }
 
        private Task Client_OnSubscribed(object Sender, ushort PacketIdentifier)
        {
            this.Model.ExternalEvent(this, "OnSubscribed",
                new KeyValuePair<string, object>("PacketIdentifier", PacketIdentifier),
                new KeyValuePair<string, object>("Client", this.client));
 
            return Task.CompletedTask;
        }
 
        private Task Client_OnPublished(object Sender, ushort PacketIdentifier)
        {
            this.Model.ExternalEvent(this, "OnPublished",
                new KeyValuePair<string, object>("PacketIdentifier", PacketIdentifier),
                new KeyValuePair<string, object>("Client", this.client));
 
            return Task.CompletedTask;
        }
 
        private Task Client_OnPingResponse(object Sender, EventArgs e)
        {
            this.Model.ExternalEvent(this, "OnPingResponse",
                new KeyValuePair<string, object>("e", e),
                new KeyValuePair<string, object>("Client", this.client));
 
            return Task.CompletedTask;
        }
 
        private Task Client_OnPing(object Sender, EventArgs e)
        {
            this.Model.ExternalEvent(this, "OnPing",
                new KeyValuePair<string, object>("e", e),
                new KeyValuePair<string, object>("Client", this.client));
 
            return Task.CompletedTask;
        }
 
        private Task Client_OnContentReceived(object Sender, MqttContent Content)
        {
            this.Model.ExternalEvent(this, "OnContentReceived",
                new KeyValuePair<string, object>("Content", Content),
                new KeyValuePair<string, object>("Client", this.client));
 
            return Task.CompletedTask;
        }
 
        private Task Client_OnError(object Sender, Exception Exception)
        {
            this.Model.ExternalEvent(this, "Error",
                new KeyValuePair<string, object>("Exception", Exception),
                new KeyValuePair<string, object>("Client", this.client));
 
            return Task.CompletedTask;
        }
 
        private Task Client_OnConnectionError(object Sender, Exception Exception)
        {
            this.Model.ExternalEvent(this, "ConnectionError",
                new KeyValuePair<string, object>("Exception", Exception),
                new KeyValuePair<string, object>("Client", this.client));
 
            return Task.CompletedTask;
        }
 
        private Task Client_OnStateChanged(object Sender, MqttState NewState)
        {
            switch (NewState)
            {
                case MqttState.Connected:
                    this.isOnline = true;
 
                    if (string.IsNullOrEmpty(this.credentials.ObjectId))
                        Database.Insert(this.credentials);
 
                    this.client.SUBSCRIBE(this.subscriptions);
 
                    this.connected?.TrySetResult(true);
                    break;
 
                case MqttState.Error:
                case MqttState.Offline:
                    this.isOnline = false;
                    this.connected?.TrySetResult(false);
                    break;
            }
 
            this.Model.ExternalEvent(this, "OnStateChanged",
                new KeyValuePair<string, object>("NewState", NewState),
                new KeyValuePair<string, object>("Client", this.client));
 
            return Task.CompletedTask;
        }
 
        public override async Task StartInstance()
        {
            if (!(await this.connected.Task))
                throw new Exception("Unable to connect " + this.userName + "@" + this.domain);
 
            this.connected = null;
        }
 
        public override Task FinalizeInstance()
        {
            this.client?.Dispose();
            this.client = null;
 
            if (!(this.sniffer is null))
            {
                if (this.sniffer is IDisposable Disposable)
                    Disposable.Dispose();
 
                this.sniffer = null;
            }
 
            return Task.CompletedTask;
        }
 
        public override object ActivityObject
        {
            get
            {
                return new MqttActivityObject()
                {
                    Client = this.client,
                    UserName = this.userName,
                    InstanceId = this.InstanceId,
                    InstanceIndex = this.InstanceIndex
                };
            }
        }
 
    }
}