common/AMWD.Common/Utilities/AsyncQueue.cs

using System.Diagnostics.CodeAnalysis;
using System.Threading;
using System.Threading.Tasks;

namespace System.Collections.Generic
{
	/// <summary>
	/// Represents a first-in, first-out collection of objects.
	/// </summary>
	/// <typeparam name="T">Specifies the type of elements in the queue.</typeparam>
	public class AsyncQueue<T>
	{
		#region Fields

		private readonly Queue<T> queue;

		private TaskCompletionSource<bool> dequeueTcs = new();
		private TaskCompletionSource<bool> availableTcs = new();

		#endregion Fields

		#region Constructors

		/// <summary>
		/// Initializes a new instance of the <see cref="AsyncQueue{T}"/> class that is empty and has the default initial capacity.
		/// </summary>
		[ExcludeFromCodeCoverage]
		public AsyncQueue()
		{
			queue = new Queue<T>();
		}

		/// <summary>
		/// Initializes a new instance of the System.Collections.Generic.Queue`1 class that
		/// contains elements copied from the specified collection and has sufficient capacity
		/// to accommodate the number of elements copied.
		/// </summary>
		/// <param name="collection">The collection whose elements are copied to the new <see cref="AsyncQueue{T}"/>.</param>
		/// <exception cref="ArgumentNullException"><paramref name="collection"/> is null.</exception>
		[ExcludeFromCodeCoverage]
		public AsyncQueue(IEnumerable<T> collection)
		{
			queue = new Queue<T>();
			Enqueue(collection);
		}

		/// <summary>
		/// Initializes a new instance of the <see cref="AsyncQueue{T}"/> class that is empty and has the specified initial capacity.
		/// </summary>
		/// <param name="capacity">The initial number of elements that the <see cref="AsyncQueue{T}"/> can contain.</param>
		/// <exception cref="ArgumentOutOfRangeException"><paramref name="capacity"/> is less than zero.</exception>
		[ExcludeFromCodeCoverage]
		public AsyncQueue(int capacity)
		{
			queue = new Queue<T>(capacity);
		}

		#endregion Constructors

		#region Properties

		/// <summary>
		/// Gets the number of elements contained in the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		[ExcludeFromCodeCoverage]
		public int Count
		{
			get
			{
				lock (queue)
				{
					return queue.Count;
				}
			}
		}

		#endregion Properties

		#region Queue implementation

		/// <summary>
		/// Removes all objects from the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		[ExcludeFromCodeCoverage]
		public void Clear()
		{
			lock (queue)
			{
				queue.Clear();
			}
		}

		/// <summary>
		/// Determines whether an element is in the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		/// <param name="item">The object to locate in the <see cref="AsyncQueue{T}"/>. The value can be null for reference types.</param>
		/// <returns>true if item is found in the <see cref="AsyncQueue{T}"/>; otherwise, false.</returns>
		[ExcludeFromCodeCoverage]
		public bool Contains(T item)
		{
			lock (queue)
			{
				return queue.Contains(item);
			}
		}

		/// <summary>
		/// Copies the <see cref="AsyncQueue{T}"/> elements to an existing one-dimensional <see cref="Array"/>, starting at the specified array index.
		/// </summary>
		/// <param name="array">The one-dimensional <see cref="Array"/> that is the destination of the elements copied from <see cref="AsyncQueue{T}"/>. The <see cref="Array"/> must have zero-based indexing.</param>
		/// <param name="arrayIndex">The zero-based index in array at which copying begins.</param>
		/// <exception cref="ArgumentNullException"><paramref name="array"/> is null.</exception>
		/// <exception cref="ArgumentOutOfRangeException"><paramref name="arrayIndex"/> is less than zero.</exception>
		/// <exception cref="ArgumentException">The number of elements in the source <see cref="AsyncQueue{T}"/> is greater than the available space from <paramref name="arrayIndex"/> to the end of the destination array.</exception>
		[ExcludeFromCodeCoverage]
		public void CopyTo(T[] array, int arrayIndex)
		{
			lock (queue)
			{
				queue.CopyTo(array, arrayIndex);
			}
		}

		/// <summary>
		/// Removes and returns the object at the beginning of the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		/// <returns>The object that is removed from the beginning of the <see cref="AsyncQueue{T}"/>.</returns>
		/// <exception cref="InvalidOperationException">The <see cref="AsyncQueue{T}"/> is empty.</exception>
		[ExcludeFromCodeCoverage]
		public T Dequeue()
		{
			lock (queue)
			{
				return queue.Dequeue();
			}
		}

		/// <summary>
		/// Adds an object to the end of the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		/// <param name="item">The object to add to the <see cref="AsyncQueue{T}"/>. The value can be null for reference types.</param>
		public void Enqueue(T item)
		{
			lock (queue)
			{
				queue.Enqueue(item);
				SetToken(dequeueTcs);
				SetToken(availableTcs);
			}
		}

		/// <summary>
		/// Returns the object at the beginning of the <see cref="AsyncQueue{T}"/> without removing it.
		/// </summary>
		/// <returns>The object at the beginning of the <see cref="AsyncQueue{T}"/>.</returns>
		/// <exception cref="InvalidOperationException">The <see cref="AsyncQueue{T}"/> is empty.</exception>
		[ExcludeFromCodeCoverage]
		public T Peek()
		{
			lock (queue)
			{
				return queue.Peek();
			}
		}

		/// <summary>
		/// Copies the <see cref="AsyncQueue{T}"/> elements to a new array.
		/// </summary>
		/// <returns>A new array containing elements copied from the <see cref="AsyncQueue{T}"/>.</returns>
		[ExcludeFromCodeCoverage]
		public T[] ToArray()
		{
			lock (queue)
			{
				return queue.ToArray();
			}
		}

		/// <summary>
		/// Sets the capacity to the actual number of elements in the <see cref="AsyncQueue{T}"/>, if that number is less than 90 percent of current capacity.
		/// </summary>
		[ExcludeFromCodeCoverage]
		public void TrimExcess()
		{
			lock (queue)
			{
				queue.TrimExcess();
			}
		}

		#endregion Queue implementation

		#region Async implementation

		/// <summary>
		/// Removes and returns all available objects at the beginning of the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		/// <param name="maxCount">The maximum number of objects to return. Zero means no limit.</param>
		/// <param name="cancellationToken">A cancellation token used to propagate notification that this operation should be canceled.</param>
		/// <returns>The objects that are removed from the beginning of the <see cref="AsyncQueue{T}"/>.</returns>
		public async Task<T[]> DequeueAvailableAsync(int maxCount = 0, CancellationToken cancellationToken = default)
		{
			while (true)
			{
				TaskCompletionSource<bool> internalDequeueTcs;
				lock (queue)
				{
					if (queue.Count > 0)
					{
						int count = queue.Count;
						if (maxCount > 0 && count > maxCount)
							count = maxCount;

						var items = new T[count];
						for (int i = 0; i < count; i++)
							items[i] = queue.Dequeue();

						return items;
					}
					internalDequeueTcs = ResetToken(ref dequeueTcs);
				}

				await WaitAsync(internalDequeueTcs, cancellationToken);
			}
		}

		/// <summary>
		/// Removes and returns objects at the beginning of the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		/// <param name="count">The number of objects to return.</param>
		/// <param name="cancellationToken">A cancellation token used to propagate notification that this operation should be canceled.</param>
		/// <returns>The objects that are removed from the beginning of the <see cref="AsyncQueue{T}"/>.</returns>
		public async Task<T[]> DequeueManyAsync(int count, CancellationToken cancellationToken = default)
		{
			if (count < 0)
				throw new ArgumentOutOfRangeException(nameof(count));

			while (true)
			{
				TaskCompletionSource<bool> internalDequeueTcs;
				lock (queue)
				{
					if (count <= queue.Count)
					{
						var items = new T[count];
						for (int i = 0; i < count; i++)
							items[i] = queue.Dequeue();

						return items;
					}
					internalDequeueTcs = ResetToken(ref dequeueTcs);
				}

				await WaitAsync(internalDequeueTcs, cancellationToken);
			}
		}

		/// <summary>
		/// Removes and returns the object at the beginning of the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		/// <param name="cancellationToken">A cancellation token used to propagate notification that this operation should be canceled.</param>
		/// <returns>The object that is removed from the beginning of the <see cref="AsyncQueue{T}"/>.</returns>
		public async Task<T> DequeueAsync(CancellationToken cancellationToken = default)
		{
			while (true)
			{
				TaskCompletionSource<bool> internalDequeueTcs;
				lock (queue)
				{
					if (queue.Count > 0)
						return queue.Dequeue();

					internalDequeueTcs = ResetToken(ref dequeueTcs);
				}

				await WaitAsync(internalDequeueTcs, cancellationToken);
			}
		}

		/// <summary>
		/// Waits asynchonously until at least one object is available in the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		/// <param name="cancellationToken">A cancellation token used to propagate notification that this operation should be canceled.</param>
		/// <returns>An awaitable task.</returns>
		public async Task WaitAsync(CancellationToken cancellationToken = default)
		{
			TaskCompletionSource<bool> internalAvailableTcs;
			lock (queue)
			{
				if (queue.Count > 0)
					return;

				internalAvailableTcs = ResetToken(ref availableTcs);
			}

			await WaitAsync(internalAvailableTcs, cancellationToken);
		}

		#endregion Async implementation

		#region Additional features

		/// <summary>
		/// Removes the object at the beginning of the <see cref="AsyncQueue{T}"/>, and copies it to the <paramref name="result"/> parameter.
		/// </summary>
		/// <param name="result">The removed object.</param>
		/// <returns>true if the object is successfully removed; false if the <see cref="AsyncQueue{T}"/> is empty.</returns>
		public bool TryDequeue(out T result)
		{
			try
			{
				result = Dequeue();
				return true;
			}
			catch
			{
				result = default;
				return false;
			}
		}

		/// <summary>
		/// Returns a value that indicates whether there is an object at the beginning of
		/// the <see cref="AsyncQueue{T}"/>, and if one is present, copies it to the
		/// <paramref name="result"/> parameter. The object is not removed from the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		/// <param name="result">If present, the object at the beginning of the <see cref="AsyncQueue{T}"/>; otherwise, the default value of <typeparamref name="T"/>.</param>
		/// <returns>true if there is an object at the beginning of the <see cref="AsyncQueue{T}"/>; false if the <see cref="AsyncQueue{T}"/> is empty.</returns>
		public bool TryPeek(out T result)
		{
			try
			{
				result = Peek();
				return true;
			}
			catch
			{
				result = default;
				return false;
			}
		}

		/// <summary>
		/// Removes the first occurrence of a specific object from the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		/// <param name="item">The object to remove from the <see cref="AsyncQueue{T}"/>. The value can be null for reference types.</param>
		/// <returns>true if item is successfully removed; otherwise, false. This method also returns false if item was not found in the <see cref="AsyncQueue{T}"/>.</returns>
		public bool Remove(T item)
		{
			lock (queue)
			{
				var copy = new Queue<T>(queue);
				queue.Clear();

				bool found = false;
				int count = copy.Count;
				for (int i = 0; i < count; i++)
				{
					var element = copy.Dequeue();
					if (found)
					{
						queue.Enqueue(element);
						continue;
					}

					if ((element == null && item == null) || element?.Equals(item) == true)
					{
						found = true;
						continue;
					}

					queue.Enqueue(element);
				}

				return found;
			}
		}

		/// <summary>
		/// Adds objects to the end of the <see cref="AsyncQueue{T}"/>.
		/// </summary>
		/// <param name="collection">The objects to add to the <see cref="AsyncQueue{T}"/>.</param>
		public void Enqueue(IEnumerable<T> collection)
		{
			lock (queue)
			{
				bool hasElements = false;
				foreach (var element in collection)
				{
					hasElements = true;
					queue.Enqueue(element);
				}

				if (hasElements)
				{
					SetToken(dequeueTcs);
					SetToken(availableTcs);
				}
			}
		}

		#endregion Additional features

		#region Helper

		[ExcludeFromCodeCoverage]
		private static void SetToken(TaskCompletionSource<bool> tcs)
		{
			tcs.TrySetResult(true);
		}

		[ExcludeFromCodeCoverage]
		private static TaskCompletionSource<bool> ResetToken(ref TaskCompletionSource<bool> tcs)
		{
			if (tcs.Task.IsCompleted)
			{
				tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
			}

			return tcs;
		}

		[ExcludeFromCodeCoverage]
		private static async Task WaitAsync(TaskCompletionSource<bool> tcs, CancellationToken cancellationToken)
		{
			if (await Task.WhenAny(tcs.Task, Task.Delay(-1, cancellationToken)) == tcs.Task)
			{
				await tcs.Task;
				return;
			}

			cancellationToken.ThrowIfCancellationRequested();
		}

		#endregion Helper
	}
}