PipeEventSink.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.IO.Pipes;
using System.Security.Principal;
using System.Text;
using System.Threading.Tasks;
using Waher.Content.Xml;
using Waher.Events.Files;
 
namespace Waher.Events.Pipe
{
    public delegate NamedPipeClientStream NamedPipeClientStreamFactory(string Name);
 
    public class PipeEventSink : EventSink
    {
        private readonly LinkedList<byte[]> pipeQueue = new LinkedList<byte[]>();
        private readonly NamedPipeClientStreamFactory pipeStreamFactory;
        private readonly string pipeName;
        private NamedPipeClientStream pipe;
        private bool writing = false;
 
        public PipeEventSink(string ObjectId, string PipeName)
            : this(ObjectId, PipeName, DefaultFactory)
        {
        }
 
        public PipeEventSink(string ObjectId, string PipeName, NamedPipeClientStreamFactory StreamFactory)
            : base(ObjectId)
        {
            this.pipeStreamFactory = StreamFactory;
            this.pipe = null;
            this.pipeName = PipeName;
        }
 
        private static NamedPipeClientStream DefaultFactory(string Name)
        {
            return new NamedPipeClientStream(".", Name, PipeDirection.Out, PipeOptions.Asynchronous,
                TokenImpersonationLevel.Anonymous, HandleInheritability.None);
        }
 
        public NamedPipeClientStream Pipe => this.pipe;
 
        public override Task DisposeAsync()
        {
            if (!(this.pipe is null))
            {
                this.pipe?.Dispose();
                this.pipe = null;
            }
 
            return base.DisposeAsync();
        }
 
        public override Task Queue(Event Event)
        {
            return this.Queue(Event.ToXML(), false);
        }
 
        public Task Queue(string Xml)
        {
            return this.Queue(Xml, true);
        }
 
        private async Task Queue(string Xml, bool ValidateXml)
        {
            if (ValidateXml && !XML.IsValidXml(Xml))
                throw new ArgumentException("Invalid XML.", nameof(Xml));
 
            try
            {
                byte[] Bin = Encoding.UTF8.GetBytes(Xml);
 
                lock (this.pipeQueue)
                {
                    if (this.writing)
                    {
                        this.pipeQueue.AddLast(Bin);
                        return;
                    }
                    else
                        this.writing = true;
                }
 
                if (!(this.pipe is null) && !this.pipe.IsConnected)
                {
                    this.pipe.Dispose();
                    this.pipe = null;
                }
 
                if (this.pipe is null)
                {
                    this.pipe = this.pipeStreamFactory(this.pipeName);
                    await this.BeforeConnect.Raise(this, EventArgs.Empty);
                    await this.pipe.ConnectAsync(5000);
                    await this.AfterConnect.Raise(this, EventArgs.Empty);
                }
 
                while (!(Bin is null))
                {
                    await this.pipe.WriteAsync(Bin, 0, Bin.Length);
 
                    lock (this.pipeQueue)
                    {
                        if (this.pipeQueue.First is null)
                        {
                            this.writing = false;
                            Bin = null;
                        }
                        else
                        {
                            Bin = this.pipeQueue.First.Value;
                            this.pipeQueue.RemoveFirst();
                        }
                    }
                }
            }
            catch (TimeoutException)
            {
                this.EmptyPipeQueue();
            }
            catch (IOException)
            {
                this.EmptyPipeQueue();
            }
            catch (Exception ex)
            {
                this.EmptyPipeQueue();
                Log.Exception(ex);
            }
        }
 
        private void EmptyPipeQueue()
        {
            lock (this.pipeQueue)
            {
                this.pipeQueue.Clear();
                this.writing = false;
            }
        }
 
        public event EventHandlerAsync BeforeConnect;
        
        public event EventHandlerAsync AfterConnect;
    }
}