AsyncQueue.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Waher.Events;
 
namespace Waher.Runtime.Queue
{
    public class AsyncQueue<T> : IDisposable
        where T : class
    {
        private readonly LinkedList<Item> queue = new LinkedList<Item>();
        private readonly LinkedList<TaskCompletionSource<T>> subscribers = new LinkedList<TaskCompletionSource<T>>();
        private readonly TaskCompletionSource<bool> terminatedTask = new TaskCompletionSource<bool>();
        private readonly object synchObj = new object();
        private volatile int countItems = 0;
        private volatile int countSubscribers = 0;
        private bool disposed = false;
        private bool terminated = false;
        private bool empty = true;
        private bool waiting = false;
 
        public AsyncQueue()
        {
        }
 
        private class Item
        {
            internal T Value { get; }
            internal TaskCompletionSource<bool> Forwarded;
 
            internal Item(T Value)
            {
                this.Value = Value;
                this.Forwarded = null;
            }
 
            internal Item(T Value, TaskCompletionSource<bool> Forwarded)
            {
                this.Value = Value;
                this.Forwarded = Forwarded;
            }
        }
 
        public int CountItems
        {
            get
            {
                lock (this.synchObj)
                {
                    return this.countItems;
                }
            }
        }
 
        public int CountSubscribers
        {
            get
            {
                lock (this.synchObj)
                {
                    return this.countSubscribers;
                }
            }
        }
 
        public bool Empty
        {
            get
            {
                lock (this.synchObj)
                {
                    return this.empty;
                }
            }
        }
 
        public bool Waiting
        {
            get
            {
                lock (this.synchObj)
                {
                    return this.waiting;
                }
            }
        }
 
        [Obsolete("Use the Forward or Queue methods instead, for increased clarity.")]
        public Task<bool> Add(T Item)
        {
            return this.Forward(Item, false);
        }
 
        [Obsolete("Use the Forward or Queue methods instead, for increased clarity.")]
        public Task<bool> AddLast(T Item)
        {
            return this.Forward(Item, false);
        }
 
        [Obsolete("Use the Forward or Queue methods instead, for increased clarity.")]
        public Task<bool> AddFirst(T Item)
        {
            return this.Forward(Item, true);
        }
 
        [Obsolete("Use the Forward or Queue methods instead, for increased clarity.")]
        public Task<bool> Add(T Item, bool First)
        {
            return this.Forward(Item, First);
        }
 
        public void Queue(T Item)
        {
            this.Queue(Item, false);
        }
 
        public void QueueLast(T Item)
        {
            this.Queue(Item, false);
        }
 
        public void QueueFirst(T Item)
        {
            this.Queue(Item, true);
        }
 
        public void Queue(T Item, bool First)
        {
            EventHandler h = null;
 
            lock (this.synchObj)
            {
                if (this.terminated || this.disposed)
                    return;
 
                if (this.subscribers.First is null)
                {
                    Item Record = new Item(Item);
 
                    if (First)
                        this.queue.AddFirst(Record);
                    else
                        this.queue.AddLast(Record);
 
                    this.countItems++;
                    if (this.empty)
                    {
                        this.empty = false;
                        h = this.OnNotEmpty;
                    }
                }
                else
                {
                    TaskCompletionSource<T> Waiter = this.subscribers.First.Value;
                    this.subscribers.RemoveFirst();
                    this.countSubscribers--;
                    if (this.countSubscribers <= 0)
                    {
                        this.waiting = false;
                        h = this.OnNotWaiting;
                    }
 
                    Task.Run(() => Waiter.TrySetResult(Item));  // Ensures waiting logic not interrupting current logic.
                }
            }
 
            h?.Raise(this, EventArgs.Empty);
        }
 
        public Task<bool> Forward(T Item)
        {
            return this.Forward(Item, false);
        }
 
        public Task<bool> ForwardLast(T Item)
        {
            return this.Forward(Item, false);
        }
 
        public Task<bool> ForwardFirst(T Item)
        {
            return this.Forward(Item, true);
        }
 
        public Task<bool> Forward(T Item, bool First)
        {
            EventHandler h = null;
            Task<bool> Result;
 
            lock (this.synchObj)
            {
                if (this.terminated || this.disposed)
                    return Task.FromResult(false);
 
                if (this.subscribers.First is null)
                {
                    TaskCompletionSource<bool> Forwarded = new TaskCompletionSource<bool>();
 
                    Item Record = new Item(Item, Forwarded);
 
                    if (First)
                        this.queue.AddFirst(Record);
                    else
                        this.queue.AddLast(Record);
 
                    this.countItems++;
                    if (this.empty)
                    {
                        this.empty = false;
                        h = this.OnNotEmpty;
                    }
 
                    Result = Record.Forwarded.Task;
                }
                else
                {
                    TaskCompletionSource<T> Waiter = this.subscribers.First.Value;
                    this.subscribers.RemoveFirst();
                    this.countSubscribers--;
                    if (this.countSubscribers <= 0)
                    {
                        this.waiting = false;
                        h = this.OnNotWaiting;
                    }
 
                    Waiter.TrySetResult(Item);
 
                    Result = Task.FromResult(true);
                }
            }
 
            h?.Raise(this, EventArgs.Empty);
 
            return Result;
        }
 
        public Task<T> Wait()
        {
            return this.DoWait(CancellationToken.None, null);
        }
 
 
        public Task<T> Wait(int Timeout)
        {
            return this.DoWait(CancellationToken.None, Timeout);
        }
 
        public Task<T> Wait(CancellationToken Cancel)
        {
            return this.DoWait(Cancel, null);
        }
 
        public Task<T> Wait(CancellationToken Cancel, int Timeout)
        {
            return this.DoWait(Cancel, Timeout);
        }
 
        private Task<T> DoWait(CancellationToken Cancel, int? Timeout)
        {
            EventHandler h = null;
            Task<T> Result;
            Item Record;
 
            if (Timeout.HasValue && Timeout.Value <= 0)
                throw new ArgumentException("Timeout must be positive.", nameof(Timeout));
 
            lock (this.synchObj)
            {
                if (this.queue.First is null)
                {
                    if (this.disposed)
                        return Task.FromResult<T>(null);
 
                    TaskCompletionSource<T> Item = new TaskCompletionSource<T>();
 
                    if (Cancel.CanBeCanceled)
                    {
                        Cancel.Register(() => Item.TrySetResult(null));
                        if (Cancel.IsCancellationRequested)
                            return Task.FromResult<T>(null);
                    }
 
                    this.subscribers.AddLast(Item);
                    this.countSubscribers++;
                    if (!this.waiting)
                    {
                        this.waiting = true;
                        h = this.OnWaiting;
                    }
 
                    Result = Item.Task;
 
                    if (Timeout.HasValue)
                    {
                        Task.Delay(Timeout.Value).ContinueWith((_) =>
                        {
                            EventHandler h2 = null;
 
                            lock (this.synchObj)
                            {
                                if (!this.disposed && this.subscribers.Remove(Item))
                                {
                                    this.countSubscribers--;
                                    if (this.countSubscribers <= 0)
                                    {
                                        this.waiting = false;
                                        h2 = this.OnNotWaiting;
                                    }
 
                                    Item.TrySetResult(null);
                                }
                            }
 
                            h2?.Raise(this, EventArgs.Empty);
 
                            return Task.CompletedTask;
                        });
                    }
                }
                else
                {
                    Record = this.queue.First.Value;
                    this.queue.RemoveFirst();
                    this.countItems--;
                    if (this.countItems <= 0)
                    {
                        this.empty = true;
                        h = this.OnEmpty;
                    }
 
                    Record.Forwarded?.TrySetResult(true);
 
                    if (this.terminated && this.queue.First is null)
                    {
                        this.disposed = true;
                        this.terminatedTask.TrySetResult(true);
                    }
                    
                    Result = Task.FromResult(Record.Value);
                }
            }
 
            h?.Raise(this, EventArgs.Empty);
 
            return Result;
        }
 
        public bool TryPeekItem(out T Item)
        {
            return this.TryGetItem(false, out Item);
        }
 
        public bool TryGetItem(out T Item)
        {
            return this.TryGetItem(true, out Item);
        }
 
        private bool TryGetItem(bool Remove, out T Item)
        {
            EventHandler h = null;
            bool Result;
 
            lock (this.synchObj)
            {
                if (this.disposed || this.queue.First is null)
                {
                    Item = null;
                    Result = false;
                }
                else
                {
                    Item Record = this.queue.First.Value;
                    Item = Record.Value;
 
                    if (Remove)
                    {
                        this.queue.RemoveFirst();
                        this.countItems--;
                        if (this.countItems <= 0)
                        {
                            this.empty = true;
                            h = this.OnEmpty;
                        }
 
                        Record.Forwarded?.TrySetResult(true);
 
                        if (this.terminated && this.queue.First is null)
                        {
                            this.disposed = true;
                            Result = false;
                        }
                        else
                            Result = true;
                    }
                    else
                        Result = true;
                }
            }
 
            h?.Raise(this, EventArgs.Empty);
 
            return Result;
        }
 
        public void Dispose()
        {
            lock (this.synchObj)
            {
                if (this.disposed)
                    return;
 
                this.disposed = true;
                this.terminated = true;
 
                foreach (Item Record in this.queue)
                    Record.Forwarded?.TrySetResult(false);
 
                this.queue.Clear();
                this.countItems = 0;
 
                foreach (TaskCompletionSource<T> Task in this.subscribers)
                    Task.TrySetResult(null);
 
                this.subscribers.Clear();
                this.countSubscribers = 0;
            }
 
            this.RaiseDisposed();
        }
 
        public Task Terminate()
        {
            lock (this.synchObj)
            {
                this.terminated = true;
 
                if (this.queue.First is null)
                {
                    this.disposed = true;
 
                    foreach (Item Record in this.queue)
                        Record.Forwarded?.TrySetResult(false);
 
                    this.queue.Clear();
                    this.countItems = 0;
 
                    foreach (TaskCompletionSource<T> Task in this.subscribers)
                        Task.TrySetResult(null);
 
                    this.subscribers.Clear();
                    this.countSubscribers = 0;
                }
                else
                    return this.terminatedTask.Task;
            }
 
            this.RaiseDisposed();
            return this.terminatedTask.Task;
        }
 
        public event EventHandler Disposed = null;
 
        public event EventHandler OnEmpty = null;
 
        public event EventHandler OnNotEmpty = null;
 
        public event EventHandler OnWaiting = null;
 
        public event EventHandler OnNotWaiting = null;
 
        private void RaiseDisposed()
        {
            try
            {
                this.terminatedTask.TrySetResult(true);
 
                this.Disposed.Raise(this, EventArgs.Empty);
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
            }
        }
    }
}