BoshSession.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Net.Security;
using System.Text;
using System.Threading.Tasks;
using System.Xml;
using Waher.Content.Xml;
using Waher.Events;
using Waher.Networking.HTTP;
using Waher.Networking.SASL;
using Waher.Networking.Sniffers;
using Waher.Runtime.Threading;
using Waher.Runtime.Timing;
using Waher.Persistence;
using Waher.Security;
 
namespace Waher.Networking.XMPP.Server.BOSH
{
    public class BoshSession : ClientConnection
    {
        private readonly BoshWebClientResource webResource;
        private HttpResponse waiting = null;
        private readonly LinkedList<OutputRec> outputQueue = new LinkedList<OutputRec>();
        private readonly LinkedList<(long, XmlElement, Stream, string)> inputQueue = new LinkedList<(long, XmlElement, Stream, string)>();
        private MultiReadSingleWriteObject syncObjOutput;
        private MultiReadSingleWriteObject syncObjInput;
        private DateTime maxRidTimestamp = DateTime.MinValue;
        private DateTime firstEmpty = DateTime.MinValue;
        private Stream currentStream = null;
        private readonly string sid;
        private readonly CaseInsensitiveString from;
        private readonly CaseInsensitiveString to;
        private readonly string language;
        private string key = null;
        private string waitingEcho = null;
        private readonly double version;
        private long? waitingRid = null;
        private long? lastRid = null;
        private long maxRid = -1;
        private readonly int hold;
        private readonly int waitSeconds;
        private readonly int pollingSeconds;
        private readonly int requests;
        private int nrEmpty = 0;
        private readonly bool ack;
        private readonly bool secure;
        private bool terminated = false;
        private string terminationCondition = null;
 
        private readonly string remoteEndpoint = string.Empty;
        private DateTime nextTimeout = DateTime.MinValue;
        private bool removed = false;
 
        public BoshSession(BoshWebClientResource WebResource, string Sid, long? Rid,
            double Version, CaseInsensitiveString From, CaseInsensitiveString To, string Language, int Hold, int WaitSeconds,
            int PollingSeconds, int Requests, bool Ack, bool Secure, string RemoteEndpoint, string Key, XmppServer Server,
            params ISniffer[] Sniffers)
            : base(Server, XmppConnectionState.StreamNegotiation, Sniffers)
        {
            this.syncObjOutput = new MultiReadSingleWriteObject(this);
            this.syncObjInput = new MultiReadSingleWriteObject(this);
 
            this.webResource = WebResource;
            this.sid = Sid;
            this.lastRid = Rid;
            this.version = Version;
            this.from = From;
            this.to = To;
            this.language = Language;
            this.hold = Hold;
            this.waitSeconds = WaitSeconds;
            this.pollingSeconds = PollingSeconds;
            this.requests = Requests;
            this.ack = Ack;
            this.secure = Secure;
            this.remoteEndpoint = RemoteEndpoint;
            this.key = Key;
        }
 
        private class OutputRec
        {
            public string Payload;
            public EventHandlerAsync<DeliveryEventArgs> Callback;
            public object State;
        }
 
        public override string Binding => "BOSH";
 
        public string Sid => this.sid;
 
        public double Version => this.version;
 
        public CaseInsensitiveString From => this.from;
 
        public CaseInsensitiveString To => this.to;
 
        public string Language => this.language;
 
        public int Hold => this.hold;
 
        public int WaitSeconds => this.waitSeconds;
 
        public int PollingSeconds => this.pollingSeconds;
 
        public int Requests => this.requests;
 
        public bool Ack => this.ack;
 
        public bool Secure => this.secure;
 
        public bool Terminated
        {
            get => this.terminated;
            internal set => this.terminated = value;
        }
 
        public bool IsBound
        {
            get => this.isBound;
            internal set => this.isBound = value;
        }
 
        public string Key
        {
            get => this.key;
            internal set => this.key = value;
        }
 
        internal bool Removed
        {
            get => this.removed;
            set => this.removed = value;
        }
 
        public override string RemoteEndpoint => this.remoteEndpoint;
 
        public override string Protocol => "XMPP (BOSH)";
 
        public override async Task DisposeAsync()
        {
            if (!this.disposed)
            {
                await this.SetState(XmppConnectionState.Offline);
 
                if (this.isBound && !(this.Server is null))
                {
                    this.isBound = false;
                    await this.ProcessFragment("<presence type=\"unavailable\" xmlns=\"jabber:client\"/>", null);
                    this.Server?.ConnectionClosed(this);
                }
 
                if (!(this.waiting is null))
                {
                    await this.Error("Connection disposed.");
                    await this.webResource.BoshError(this.waiting, string.Empty);
                    this.waiting = null;
                    this.waitingRid = null;
                    this.waitingEcho = null;
                }
 
                this.terminated = true;
 
                if (!this.removed)
                {
                    this.removed = true;
                    BoshWebClientResource.Remove(this);
                }
 
                this.syncObjOutput?.Dispose();
                this.syncObjOutput = null;
 
                this.syncObjInput?.Dispose();
                this.syncObjInput = null;
 
                await base.DisposeAsync();
            }
        }
 
        internal async Task RegisterForResponse(Scheduler Scheduler, long? Rid, HttpResponse Response, string Echo)
        {
            if (this.terminated)
            {
                await this.Error(this.terminationCondition);
                await this.webResource.BoshError(Response, this.terminationCondition);
                return;
            }
 
            LinkedList<OutputRec> ToCall = null;
            StringBuilder Xml = null;
            bool RemoveTimeout = false;
            bool ScheduleTimeout = false;
 
            if (!await this.syncObjOutput.TryBeginWrite(10000))
                throw new TimeoutException("Unable to get access to session.");
 
            try
            {
                if (Rid.HasValue && Rid.Value <= this.maxRid)
                {
                    await this.EmptyResponseLocked(Response, Rid, Echo);
                    return;
                }
 
                if (!(this.waiting is null))
                {
                    if (this.waitingRid.HasValue && Rid.HasValue && Rid.Value < this.waitingRid.Value)
                    {
                        await this.EmptyResponseLocked(Response, Rid, Echo);
                        return;
                    }
 
                    try
                    {
                        await this.EmptyResponseLocked(this.waiting, this.waitingRid, this.waitingEcho);
                    }
                    catch (Exception)
                    {
                        // Ignore
                    }
 
                    RemoveTimeout = true;
                    this.waiting = null;
                    this.waitingRid = null;
                    this.waitingEcho = null;
                }
 
                if (this.outputQueue.First is null)
                {
                    this.waitingRid = Rid;
                    this.waiting = Response;
                    this.waitingEcho = Echo;
                    ScheduleTimeout = true;
                }
                else
                {
                    if (Xml is null)
                        Xml = new StringBuilder();
                    else
                        Xml.Clear();
 
                    Xml.Append("<body from='");
                    Xml.Append(XML.Encode(this.to));
 
                    if (Rid.HasValue)
                    {
                        Xml.Append("' ack='");
                        Xml.Append(Rid.Value.ToString());
                        this.RidReturnedLocked(Rid.Value);
                    }
 
                    if (!string.IsNullOrEmpty(Echo))
                    {
                        Xml.Append("' echo='");
                        Xml.Append(XML.Encode(Echo));
                    }
 
                    Xml.Append("' xmlns='");
                    Xml.Append(BoshWebClientResource.HttpBindNamespace);
 
                    if (this.State != XmppConnectionState.Active)
                    {
                        Xml.Append("' xmlns:stream='");
                        Xml.Append(StreamNamespace);
                    }
 
                    Xml.Append("'>");
 
                    foreach (OutputRec Fragment in this.outputQueue)
                    {
                        Xml.Append(Fragment.Payload);
 
                        if (!(Fragment.Callback is null))
                        {
                            if (ToCall is null)
                                ToCall = new LinkedList<OutputRec>();
 
                            ToCall.AddLast(Fragment);
                        }
                    }
 
                    Xml.Append("</body>");
 
                    this.outputQueue.Clear();
 
                    string Tx = Xml.ToString();
                    await this.webResource.Return(Response, Tx);
                    await this.TransmitText(Tx);
                }
            }
            finally
            {
                await this.syncObjOutput.EndWrite();
            }
 
            if ((RemoveTimeout || ScheduleTimeout) && this.nextTimeout != DateTime.MinValue)
            {
                Scheduler.Remove(this.nextTimeout);
                this.nextTimeout = DateTime.MinValue;
            }
 
            if (ScheduleTimeout)
                this.nextTimeout = Scheduler.Add(DateTime.Now.AddSeconds(this.waitSeconds), this.Timeout, null);
 
            if (!(ToCall is null))
            {
                foreach (OutputRec Rec in ToCall)
                    await Rec.Callback.Raise(this, new DeliveryEventArgs(Rec.State, true));
            }
        }
 
        private async Task EmptyResponseLocked(HttpResponse Response, long? Rid, string Echo)
        {
            StringBuilder Xml = new StringBuilder();
 
            Xml.Append("<body from='");
            Xml.Append(XML.Encode(this.to));
 
            if (Rid.HasValue)
            {
                Xml.Append("' ack='");
                Xml.Append(Rid.ToString());
                this.RidReturnedLocked(Rid.Value);
            }
 
            if (!string.IsNullOrEmpty(Echo))
            {
                Xml.Append("' echo='");
                Xml.Append(XML.Encode(Echo));
            }
 
            Xml.Append("' xmlns='");
            Xml.Append(BoshWebClientResource.HttpBindNamespace);
            Xml.Append("'/>");
 
            string Tx = Xml.ToString();
            await this.webResource.Return(Response, Tx);
            await this.TransmitText(Tx);
        }
 
        private async void Timeout(object P)
        {
            try
            {
                StringBuilder Xml;
                HttpResponse Response;
                LinkedList<OutputRec> ToCall = null;
                long? Rid;
                string Echo;
 
                if (this.disposed || this.nextTimeout == DateTime.MinValue)
                    return;
 
                this.nextTimeout = DateTime.MinValue;
 
                try
                {
                    if (!await this.syncObjOutput.TryBeginWrite(10000))
                        throw new TimeoutException("Unable to get access to session.");
 
                    try
                    {
                        if (this.waiting is null)
                            return;
 
                        Response = this.waiting;
                        Rid = this.waitingRid;
                        Echo = this.waitingEcho;
                        this.waiting = null;
                        this.waitingRid = null;
                        this.waitingEcho = null;
 
                        Xml = new StringBuilder();
 
                        Xml.Append("<body from='");
                        Xml.Append(XML.Encode(this.to));
 
                        if (Rid.HasValue)
                        {
                            Xml.Append("' ack='");
                            Xml.Append(Rid.Value.ToString());
                            this.RidReturnedLocked(Rid.Value);
                        }
 
                        if (!string.IsNullOrEmpty(Echo))
                        {
                            Xml.Append("' echo='");
                            Xml.Append(XML.Encode(Echo));
                        }
 
                        Xml.Append("' xmlns='");
                        Xml.Append(BoshWebClientResource.HttpBindNamespace);
 
                        if (this.State != XmppConnectionState.Active)
                        {
                            Xml.Append("' xmlns:stream='");
                            Xml.Append(XmppClientConnection.StreamNamespace);
                        }
 
                        if (this.outputQueue.First is null)
                            Xml.Append("'/>");
                        else
                        {
                            Xml.Append("'>");
 
                            foreach (OutputRec Fragment in this.outputQueue)
                            {
                                Xml.Append(Fragment);
 
                                if (!(Fragment.Callback is null))
                                {
                                    if (!(ToCall is null))
                                        ToCall = new LinkedList<OutputRec>();
 
                                    ToCall.AddLast(Fragment);
                                }
                            }
 
                            Xml.Append("</body>");
 
                            this.outputQueue.Clear();
                        }
                    }
                    finally
                    {
                        await this.syncObjOutput.EndWrite();
                    }
 
                    string Tx = Xml.ToString();
                    await this.webResource.Return(Response, Tx);
                    await this.TransmitText(Tx);
                }
                catch (Exception ex)
                {
                    Log.Exception(ex);
                    // Ignore
                }
 
                if (!(ToCall is null))
                {
                    foreach (OutputRec Rec in ToCall)
                        await Rec.Callback.Raise(this, new DeliveryEventArgs(Rec.State, false));
                }
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
            }
        }
 
 
        public override async Task<bool> BeginWrite(string Xml, EventHandlerAsync<DeliveryEventArgs> Callback, object State)
        {
            if (this.disposed)
                return false;
 
            if (string.IsNullOrEmpty(Xml))
                await Callback.Raise(this, new DeliveryEventArgs(State, true));
            else
            {
                HttpResponse Response = null;
                DateTime Now = DateTime.Now;
                long? Rid;
                string Echo;
 
                this.nrEmpty = 0;
 
                if (!await this.syncObjOutput.TryBeginWrite(10000))
                    throw new TimeoutException("Unable to get access to session.");
 
                try
                {
                    if (this.waiting is null)
                    {
                        this.outputQueue.AddLast(new OutputRec()
                        {
                            Payload = Xml,
                            Callback = Callback,
                            State = State
                        });
                        return true;
                    }
 
                    Response = this.waiting;
                    Rid = this.waitingRid;
                    Echo = this.waitingEcho;
                    this.waiting = null;
                    this.waitingRid = null;
                    this.waitingEcho = null;
                }
                finally
                {
                    await this.syncObjOutput.EndWrite();
                }
 
                try
                {
                    StringBuilder Xml2 = new StringBuilder();
 
                    Xml2.Append("<body from='");
                    Xml2.Append(XML.Encode(this.to));
 
                    if (Rid.HasValue)
                    {
                        Xml2.Append("' ack='");
                        Xml2.Append(Rid.Value.ToString());
                        this.RidReturnedLocked(Rid.Value);
                    }
 
                    if (!string.IsNullOrEmpty(Echo))
                    {
                        Xml2.Append("' echo='");
                        Xml2.Append(XML.Encode(Echo));
                    }
 
                    Xml2.Append("' xmlns='");
                    Xml2.Append(BoshWebClientResource.HttpBindNamespace);
 
                    if (this.State != XmppConnectionState.Active)
                    {
                        Xml2.Append("' xmlns:stream='");
                        Xml2.Append(XmppClientConnection.StreamNamespace);
                    }
 
                    Xml2.Append("'>");
                    Xml2.Append(Xml);
                    Xml2.Append("</body>");
 
                    string Tx = Xml2.ToString();
                    await this.webResource.Return(Response, Tx);
                    await this.TransmitText(Tx);
                }
                catch (Exception)
                {
                    // Ignore
 
                    return false;
                }
 
                await Callback.Raise(this, new DeliveryEventArgs(State, true));
            }
 
            return true;
        }
 
        public override Task<bool> StreamError(string ErrorXml, string Reason)
        {
            return this.ToError("<stream:error>" + ErrorXml + "</stream:error>", Reason);
        }
 
        private async Task<bool> ToError(string ErrorXml, string Reason)
        {
            this.terminated = true;
            this.removed = true;
            BoshWebClientResource.Remove(this);
 
            Log.Error("Terminating session. " + Reason, this.FullJid);
 
            if (string.IsNullOrEmpty(ErrorXml))
            {
                await this.SetState(XmppConnectionState.Error);
                return false;
            }
            else
            {
                return await this.BeginWrite(ErrorXml, async (Sender, e) =>
                {
                    await this.SetState(XmppConnectionState.Error);
                }, null);
            }
        }
 
        private async Task<bool> ProcessFragment(string Xml, Stream Stream)
        {
            XmlDocument Doc;
 
            try
            {
                if (!string.IsNullOrEmpty(this.fullJid))
                    this.server.TouchClientConnection(this.fullJid);
 
                Doc = new XmlDocument()
                {
                    PreserveWhitespace = true
                };
                Doc.LoadXml("<body xmlns='" + BoshWebClientResource.HttpBindNamespace + "'>" + Xml + "</body>");
 
                return await this.ProcessStanzas(Doc.DocumentElement, Stream);
            }
            catch (Exception ex)
            {
                await this.Exception(ex);
                await this.StreamError("<bad-format xmlns='urn:ietf:params:xml:ns:xmpp-streams'/>", ex.Message);
 
                return false;
            }
        }
 
        internal async Task<bool> ProcessStanzasInOrder(long Rid, XmlElement RootElement, Stream Stream, string Key)
        {
            bool Process = false;
 
            if (this.terminated)
                return false;
 
            if (!await this.syncObjInput.TryBeginWrite(10000))
                throw new TimeoutException("Unable to get access to session.");
 
            try
            {
                if (!this.lastRid.HasValue || Rid == this.lastRid.Value + 1)
                {
                    if (!this.CheckNextKey(Key))
                        return false;
 
                    this.lastRid = Rid;
                    Process = true;
                }
                else
                {
                    if (this.inputQueue.Count >= this.hold || (Rid - this.lastRid.Value) > this.requests)
                    {
                        this.terminated = true;
                        this.terminationCondition = "item-not-found";
                    }
                    else
                        this.inputQueue.AddLast((Rid, RootElement, Stream, Key));
 
                    Process = false;
                }
            }
            finally
            {
                await this.syncObjInput.EndWrite();
            }
 
            if (this.terminated)
                Log.Error("Terminating session. Too many incoming requests.", this.fullJid);
            else if (Process)
            {
                bool FirstElement = true;
 
                do
                {
                    if (FirstElement)
                        FirstElement = false;
 
                    XmlElement RootElement0 = RootElement;
                    Stream Stream0 = Stream;
 
                    LinkedListNode<(long, XmlElement, Stream, string)> First;
                    (long, XmlElement, Stream, string) Rec;
 
                    if (!await this.syncObjInput.TryBeginWrite(10000))
                        throw new TimeoutException("Unable to get access to session.");
 
                    try
                    {
                        if (!((First = this.inputQueue.First) is null) &&
                            (Rec = First.Value).Item1 == Rid + 1)
                        {
                            do
                            {
                                RootElement = Rec.Item2;
                                Stream = Rec.Item3;
                                Key = Rec.Item4;
                                this.inputQueue.RemoveFirst();
 
                                if (this.CheckNextKey(Key))
                                    this.lastRid = Rid;
                                else
                                    RootElement = null;
                            }
                            while (RootElement is null &&
                                !((First = this.inputQueue.First) is null) &&
                                (Rec = First.Value).Item1 == Rid + 1);
 
                            if (!(RootElement is null))
                                Rid++;
                        }
                        else
                            RootElement = null;
                    }
                    finally
                    {
                        await this.syncObjInput.EndWrite();
                    }
 
                    try
                    {
                        if (!await this.ProcessStanzas(RootElement0, Stream0))
                            return false;
                    }
                    catch (Exception ex)
                    {
                        await this.Exception(ex);
                    }
                }
                while (!(RootElement is null));
            }
 
            return true;
        }
 
        internal bool CheckNextKey(string Key)
        {
            if (Key is null)
            {
                if (!(this.key is null))
                    return false;
            }
            else
            {
                string PrevKey = Hashes.ComputeSHA1HashString(Encoding.ASCII.GetBytes(Key));
                int i, c = PrevKey.Length;
 
                if (c != this.key.Length)
                    return false;
 
                for (i = 0; i < c; i++)
                {
                    if (PrevKey[i] != this.key[i])
                        return false;
                }
 
                this.key = Key;
            }
 
            return true;
        }
 
        protected override SslStream GetSslStream()
        {
            return this.currentStream as SslStream;
        }
 
        internal async Task<bool> ProcessStanzas(XmlElement RootElement, Stream Stream)
        {
            try
            {
                if (this.HasSniffers)
                    await this.ReceiveText(RootElement.OuterXml);
 
                int Count = 0;
 
                foreach (XmlNode N in RootElement.ChildNodes)
                {
                    if (N is XmlElement E)
                    {
                        Count++;
 
                        try
                        {
                            Stanza Stanza = new Stanza(RootElement, E, RootElement.InnerXml);
 
                            this.currentStream = Stream;
                            if (!await this.ProcessStanza(Stanza))
                                return false;
                        }
                        catch (Exception ex)
                        {
                            await this.Exception(ex);
                            if (!await this.StreamError("<bad-format xmlns='urn:ietf:params:xml:ns:xmpp-streams'/>", ex.Message))
                                return false;
                        }
                    }
                }
 
                if (Count == 0)
                {
                    if (this.State != XmppConnectionState.Active)
                    {
                        StringBuilder Xml = new StringBuilder();
                        await this.InitStream(Xml, Stream as SslStream);
                        if (!await this.BeginWrite(Xml.ToString(), null, null))
                            return false;
                    }
                    else
                    {
                        DateTime Now = DateTime.Now;
 
                        if (this.nrEmpty == 0)
                            this.firstEmpty = Now;
 
                        this.nrEmpty++;
                        if (this.nrEmpty > 5)
                        {
                            double SecondsPerRequest = (Now - this.firstEmpty).TotalSeconds / this.nrEmpty;
 
                            if (SecondsPerRequest * 2 < this.pollingSeconds)
                            {
                                this.terminated = true;
                                this.terminationCondition = "policy-violation";
                                BoshWebClientResource.Remove(this);
 
                                Log.Error("Terminating session. Polling too quickly.", this.fullJid);
                            }
                        }
                    }
                }
                else
                    this.nrEmpty = 0;
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
            }
 
            return true;
        }
 
        protected override async Task<bool> ProcessBindingSpecificStanza(Stanza Stanza, XmlElement StanzaElement)
        {
            return await this.StreamError("<unsupported-stanza-type xmlns='urn:ietf:params:xml:ns:xmpp-streams'/>",
                "Unrecognized stanza: " + StanzaElement.LocalName);
        }
 
        internal async Task InitStream(StringBuilder sb, SslStream SslStream)
        {
            sb.Append("<stream:features>");
 
            if (!this.isAuthenticated)
            {
                await this.SetState(XmppConnectionState.Authenticating);
 
                sb.Append("<mechanisms xmlns='" + XmppServer.SaslNamespace + "'>");
 
                foreach (IAuthenticationMechanism Mechanism in XmppServer.mechanisms)
                {
                    if (Mechanism.Allowed(SslStream))
                    {
                        sb.Append("<mechanism>");
                        sb.Append(Mechanism.Name);
                        sb.Append("</mechanism>");
                    }
                }
 
                sb.Append("</mechanisms>");
 
                if (await this.server.CanRegister(this))
                    sb.Append("<register xmlns='http://jabber.org/features/iq-register'/>");
            }
            else if (!this.isBound)
            {
                await this.SetState(XmppConnectionState.Binding);
 
                sb.Append("<bind xmlns='" + XmppClientConnection.BindNamespace + "'/>");
                sb.Append("<session xmlns='urn:ietf:params:xml:ns:xmpp-session'/>");
            }
 
            sb.Append("</stream:features>");
        }
 
        internal void RidReturnedLocked(long Rid)
        {
            if (Rid > this.maxRid)
            {
                this.maxRid = Rid;
                this.maxRidTimestamp = DateTime.Now;
            }
        }
 
        public override bool CheckLive()
        {
            return !this.disposed && !this.removed && this.State != XmppConnectionState.Error && this.State != XmppConnectionState.Offline;
        }
 
    }
}