RoomSource.cs

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using Waher.Content.Markdown.Consolidation;
using Waher.IoTGateway;
using Waher.Persistence;
using Waher.Script;
using Waher.Script.Abstraction.Elements;
using Waher.Script.Model;
using Waher.Script.Persistence.SQL;
using Waher.Script.Persistence.SQL.Enumerators;
 
namespace Waher.Service.IoTBroker.Admin.Muc
{
    public class RoomSource : IDataSource
    {
        private readonly MucRoom room;
        private readonly string sourceName;
        private readonly bool containsWhiteSpace;
 
        public RoomSource(MucRoom Room, string SourceName)
        {
            this.room = Room;
            this.sourceName = SourceName;
 
            this.containsWhiteSpace = false;
 
            foreach (char ch in SourceName)
            {
                if (char.IsWhiteSpace(ch))
                {
                    this.containsWhiteSpace = true;
                    break;
                }
            }
        }
 
        public string CollectionName => throw InvalidOperation();
 
        public string TypeName => throw InvalidOperation();
 
        public string Name => string.Empty;
 
        private static Exception InvalidOperation()
        {
            return new InvalidOperationException("Operation not permitted on MUC room sources.");
        }
 
        public bool IsSource(string Name)
        {
            return Name == this.room.RoomId || Name == this.room.Jid;
        }
 
        public Task<bool> IsLabel(string Label)
        {
            return Task.FromResult(false);
        }
 
        public Task<IResultSetEnumerator> Find(int Offset, int Top, bool Generic, ScriptNode Where, Variables Variables,
            KeyValuePair<VariableReference, bool>[] Order, ScriptNode Node)
        {
            StringBuilder sql = new StringBuilder();
 
            sql.Append("SELECT");
 
            if (Top < int.MaxValue)
            {
                sql.Append(" TOP ");
                sql.Append(Top.ToString());
            }
 
            sql.Append(" GENERIC * FROM ");
 
            if (this.containsWhiteSpace)
            {
                sql.Append('"');
                sql.Append(this.sourceName.Replace("\"", "\\\""));
                sql.Append('"');
            }
            else
                sql.Append(this.sourceName.Replace("\"", "\\\""));
 
            if (!(Where is null))
            {
                sql.Append(" WHERE ");
                sql.Append(Where.SubExpression);
            }
 
            if (!(Order is null))
            {
                bool First = true;
 
                foreach (KeyValuePair<VariableReference, bool> P in Order)
                {
                    if (First)
                    {
                        First = false;
                        sql.Append(" ORDER BY ");
                    }
                    else
                        sql.Append(", ");
 
                    sql.Append(P.Key.VariableName);
 
                    if (!P.Value)
                        sql.Append(" DESC");
                }
            }
 
            if (Offset > 0)
            {
                sql.Append(" OFFSET ");
                sql.Append(Offset.ToString());
            }
 
            ConsolidationState State = new ConsolidationState()
            {
                ThreadId = Guid.NewGuid().ToString(),
                N = XmppServerModule.Instance.GetNrOccupants(this.room.Jid, this.room.RoomId, this.room.Domain, true, true)
            };
            ScriptConsolidator Consolidator = new ScriptConsolidator(State.ThreadId, State.N < 12 ? 12 : State.N)
            {
                Tag = State
            };
 
            Consolidator.Added += this.Consolidator_Updated;
            Consolidator.Updated += this.Consolidator_Updated;
            Consolidator.Disposed += this.Consolidator_Disposed;
 
            XmppServerModule.Instance.RegisterConsolidator(State.ThreadId, Consolidator);
 
            Gateway.XmppClient.SendMessage(Networking.XMPP.MessageType.GroupChat, this.room.Jid, string.Empty, sql.ToString(),
                string.Empty, string.Empty, State.ThreadId, string.Empty);
 
            return State.CompletionSource.Task;
        }
 
        private class ConsolidationState
        {
            public TaskCompletionSource<IResultSetEnumerator> CompletionSource = new TaskCompletionSource<IResultSetEnumerator>();
            public string ThreadId;
            public int N;
            public bool Reported = false;
        }
 
        private Task Consolidator_Disposed(object Sender, EventArgs e)
        {
            if (Sender is ScriptConsolidator Consolidator &&
                Consolidator.Tag is ConsolidationState State)
            {
                this.Report(Consolidator, State);
            }
 
            return Task.CompletedTask;
        }
 
        private async Task Consolidator_Updated(object Sender, SourceEventArgs e)
        {
            if (Sender is ScriptConsolidator Consolidator &&
                Consolidator.Tag is ConsolidationState State &&
                await Consolidator.GetNrReportedSources() == State.N)
            {
                this.Report(Consolidator, State);
            }
        }
 
        private void Report(ScriptConsolidator Consolidator, ConsolidationState State)
        {
            if (!State.Reported)
            {
                State.Reported = true;
                IVector V = Consolidator.GetResult();
                State.CompletionSource.TrySetResult(new SynchEnumerator(V.VectorElements.GetEnumerator()));
            }
        }
 
        public Task CreateIndex(string Name, string[] Fields)
        {
            throw InvalidOperation();
        }
 
        public Task DropCollection()
        {
            throw InvalidOperation();
        }
 
        public Task<bool> DropIndex(string Name)
        {
            throw InvalidOperation();
        }
 
        public Task<int?> FindDelete(bool Lazy, int Offset, int Top, ScriptNode Where, Variables Variables, KeyValuePair<VariableReference, bool>[] Order, ScriptNode Node)
        {
            throw InvalidOperation();
        }
 
        public Task Insert(bool Lazy, object Object)
        {
            throw InvalidOperation();
        }
 
        public Task Update(bool Lazy, IEnumerable<object> Objects)
        {
            throw InvalidOperation();
        }
 
        public async Task<bool> Process(IProcessor<object> Processor, int Offset, int Top, bool Generic,
            ScriptNode Where, Variables Variables, KeyValuePair<VariableReference, bool>[] Order,
            ScriptNode Node)
        {
            IResultSetEnumerator e = await this.Find(Offset, Top, Generic, Where, Variables, Order, Node);
            bool IsAsyncronous = Processor.IsAsynchronous;
 
            while (await e.MoveNextAsync())
            {
                if (IsAsyncronous)
                {
                    if (!await Processor.ProcessAsync(e.Current))
                        return false;
                }
                else
                {
                    if (!Processor.Process(e.Current))
                        return false;
                }
            }
 
            return true;
        }
 
        public Task<int?> Delete(bool Lazy, int Offset, int Top, ScriptNode Where, Variables Variables,
            KeyValuePair<VariableReference, bool>[] Order, ScriptNode Node)
        {
            throw InvalidOperation();
        }
    }
}