ChunkCounter.cs

using System;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading.Tasks;
using Waher.Networking;
using Waher.Runtime.Cache;
 
namespace Waher.Service.NeuroLedger
{
    internal class ChunkCounter : IDisposable
    {
        private readonly Cache<string, Rec> streams;
        private readonly CommunicationLayer comLayer;
        private readonly bool receiving;
        private readonly bool cacheE2e;
 
        public ChunkCounter(bool Receiving, CommunicationLayer ComLayer, bool CacheE2e)
        {
            this.receiving = Receiving;
            this.comLayer = ComLayer;
            this.cacheE2e = CacheE2e;
 
            this.streams = new Cache<string, Rec>(int.MaxValue, new TimeSpan(0, 0, 10), new TimeSpan(0, 0, 2), true);
            this.streams.Removed += this.Streams_Removed;
        }
 
        public Task Client_OnDisposed(object Sender, EventArgs e)
        {
            this.Dispose();
            return Task.CompletedTask;
        }
 
        public void Dispose()
        {
        }
 
        private class Rec
        {
            public string StreamId;
            public int NrChunks = 0;
            public long NrBytes = 0;
            public bool Complete = false;
            public bool E2e;
 
            public Rec(string StreamId, bool E2e)
            {
                this.StreamId = StreamId;
                this.E2e = E2e;
            }
        }
 
        private static readonly Regex chunk = new Regex("<(?'Tag'chunk)\\s+(((nr=['\"](?'Nr'\\d+)['\"])|(streamId=['\"](?'StreamId'[^'\"]+)['\"])|(last=['\"](?'Last'(true|false))['\"])|(xmlns=['\"](?'Namespace'[^'\"]+)['\"]))\\s*)+>(?'Base64'[^<]*)</chunk>", RegexOptions.Compiled | RegexOptions.Singleline);
 
        public Task<string> RemoveChunkMessage(string Text)
        {
            if (Text == "ChunkReceived")
                Text = string.Empty;
 
            return Task.FromResult(Text);
        }
 
        public Task<string> ShortenChunks(string Text)
        {
            Match M;
 
            if (Text.IndexOf("<chunk") >= 0 && (M = chunk.Match(Text)).Success)
            {
                string Tag = M.Groups["Tag"].Value;
                string Namespace = M.Groups["Namespace"].Value;
 
                if (Tag == "chunk")
                {
                    if (Namespace != "urn:xmpp:http")
                        return Task.FromResult(Text);
                }
                else
                    return Task.FromResult(Text);
 
                string StreamId = M.Groups["StreamId"].Value;
                bool Last = M.Groups["Last"].Value == "true";
                string Base64 = M.Groups["Base64"].Value;
                byte[] Data;
 
                try
                {
                    Data = Convert.FromBase64String(Base64);
                }
                catch (Exception)
                {
                    return Task.FromResult(Text);
                }
 
                int NrBytes = Data.Length;
 
                if (!this.streams.TryGetValue(StreamId, out Rec Rec))
                {
                    Rec = new Rec(StreamId, false);
                    this.streams.Add(StreamId, Rec);
                }
 
                Rec.NrChunks++;
                Rec.NrBytes += NrBytes;
 
                if (Last)
                {
                    Rec.Complete = true;
                    this.streams.Remove(StreamId);
                }
 
                Text = string.Empty;
                return Task.FromResult(Text);
            }
 
            int i = Text.IndexOf("<resp ");
            if (i >= 0)
            {
                i = Text.IndexOf("<data><base64>", i + 6);
                if (i < 0)
                    return Task.FromResult(Text);
 
                i += 14;
 
                int j = Text.IndexOf("</base64></data>", i);
                if (j < 0)
                    return Task.FromResult(Text);
 
                string Base64 = Text.Substring(i, j - i);
                byte[] Data;
 
                try
                {
                    Data = Convert.FromBase64String(Base64);
                }
                catch (Exception)
                {
                    return Task.FromResult(Text);
                }
 
                Text = Text.Substring(0, i) + "(" + Data.Length.ToString() + " bytes)" + Text.Substring(j);
            }
 
            return Task.FromResult(Text);
        }
 
        private async Task Streams_Removed(object Sender, CacheItemEventArgs<string, Rec> e)
        {
            if (this.comLayer.HasSniffers)
            {
                StringBuilder sb = new StringBuilder();
                Rec Rec = e.Value;
 
                if (Rec.E2e)
                {
                    sb.Append("(");
                    sb.Append(Rec.NrChunks.ToString());
                    sb.Append(" stanzas containing ");
                    sb.Append(Rec.NrBytes.ToString());
                    sb.Append(" cipher bytes");
 
                    if (this.receiving)
                        sb.Append(" received from ");
                    else
                        sb.Append(" transmitted to ");
 
                    sb.Append(Rec.StreamId);
                    sb.Append(".)");
                }
                else
                {
                    sb.Append("(");
                    sb.Append(Rec.NrChunks.ToString());
                    sb.Append(" chunks containing ");
                    sb.Append(Rec.NrBytes.ToString());
                    sb.Append(" bytes pertaining to stream ");
                    sb.Append(Rec.StreamId);
 
                    if (this.receiving)
                        sb.Append(" received.");
                    else
                        sb.Append(" transmitted.");
 
                    if (e.Value.Complete)
                        sb.Append(" Transfer completed.");
 
                    sb.Append(")");
                }
 
                if (this.comLayer.HasSniffers)
                    await this.comLayer.Information(sb.ToString());
            }
        }
 
    }
}