AsyncQueue.cs

using System;
using System.Collections.Generic;
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;
 
        public AsyncQueue()
        {
        }
 
        private class Item
        {
            internal T Value { get; }
            internal TaskCompletionSource<bool> Processed = new TaskCompletionSource<bool>();
 
            internal Item(T Value)
            {
                this.Value = Value;
            }
        }
 
        public int CountItems
        {
            get
            {
                lock (this.synchObj)
                {
                    return this.countItems;
                }
            }
        }
 
        public int CountSubscribers
        {
            get
            {
                lock (this.synchObj)
                {
                    return this.countSubscribers;
                }
            }
        }
 
        public Task<bool> Add(T Item)
        {
            return this.Add(Item, false);
        }
 
        public Task<bool> AddLast(T Item)
        {
            return this.Add(Item, false);
        }
 
        public Task<bool> AddFirst(T Item)
        {
            return this.Add(Item, true);
        }
 
        public Task<bool> Add(T Item, bool First)
        {
            lock (this.synchObj)
            {
                if (this.terminated || this.disposed)
                    throw new ObjectDisposedException("Queue has been terminated.");
 
                if (this.subscribers.First is null)
                {
                    Item Record = new Item(Item);
 
                    if (First)
                        this.queue.AddFirst(Record);
                    else
                        this.queue.AddLast(Record);
 
                    this.countItems++;
 
                    return Record.Processed.Task;
                }
                else
                {
                    TaskCompletionSource<T> Waiter = this.subscribers.First.Value;
                    this.subscribers.RemoveFirst();
                    this.countSubscribers--;
                    Waiter.TrySetResult(Item);
 
                    return Task.FromResult(true);
                }
            }
        }
 
        public Task<T> Wait()
        {
            Item Record;
 
            lock (this.synchObj)
            {
                if (this.disposed)
                    return Task.FromResult<T>(null);
 
                if (this.queue.First is null)
                {
                    TaskCompletionSource<T> Task = new TaskCompletionSource<T>();
                    this.subscribers.AddLast(Task);
                    this.countSubscribers++;
                    return Task.Task;
                }
                else
                {
                    Record = this.queue.First.Value;
                    this.queue.RemoveFirst();
                    this.countItems--;
 
                    Record.Processed.TrySetResult(true);
 
                    if (this.terminated && this.queue.First is null)
                        this.disposed = true;
                    else
                        return Task.FromResult(Record.Value);
                }
            }
 
            this.RaiseDisposed();
 
            return Task.FromResult(Record.Value);
        }
 
        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)
        {
            lock (this.synchObj)
            {
                if (this.disposed || this.queue.First is null)
                {
                    Item = null;
                    return false;
                }
                else
                {
                    Item Record = this.queue.First.Value;
                    Item = Record.Value;
 
                    if (Remove)
                    {
                        this.queue.RemoveFirst();
                        this.countItems--;
 
                        Record.Processed.TrySetResult(true);
 
                        if (this.terminated && this.queue.First is null)
                            this.disposed = true;
                        else
                            return true;
                    }
                    else
                        return true;
                }
            }
 
            this.RaiseDisposed();
            return true;
        }
 
        public void Dispose()
        {
            lock (this.synchObj)
            {
                if (this.disposed)
                    return;
 
                this.disposed = true;
                this.terminated = true;
 
                foreach (Item Record in this.queue)
                    Record.Processed.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.Processed.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;
 
        private void RaiseDisposed()
        {
            this.terminatedTask.TrySetResult(true);
 
            try
            {
                EventHandler h = this.Disposed;
 
                if (!(h is null))
                    h(this, EventArgs.Empty);
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
            }
        }
 
    }
}