Wait until a BlockingCollection queue is cleared by a background thread, with a timeout if it takes too long?

Question

In C#, I'm wondering if it's possible to wait until a BlockingCollection is cleared by a background thread, with a timeout if it takes too long.

The temporary code that I have at the moment strikes me as somewhat inelegant (since when is it good practice to use Thread.Sleep?):

while (_blockingCollection.Count > 0 || !_blockingCollection.IsAddingCompleted)
{
    Thread.Sleep(TimeSpan.FromMilliseconds(20));
    // [extra code to break if it takes too long]
}

Answer 1

You can use GetConsumingEnumerable() and foreach in the consuming thread to determine when the queue is empty, and then set a ManualResetEvent which the main thread can check to see if the queue is empty. GetConsumingEnumerable() returns an enumerator which checks whether CompleteAdding() has been called before it terminates on an empty queue.

Sample code:

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

namespace Demo
{
    internal class Program
    {
        private void run()
        {
            Task.Run(new Action(producer));
            Task.Run(new Action(consumer));

            while (!_empty.WaitOne(1000))
                Console.WriteLine("Waiting for queue to empty");

            Console.WriteLine("Queue emptied.");
        }

        private void producer()
        {
            for (int i = 0; i < 20; ++i)
            {
                _queue.Add(i);
                Console.WriteLine("Produced " + i);
                Thread.Sleep(100);
            }

            _queue.CompleteAdding();
        }

        private void consumer()
        {
            foreach (int n in _queue.GetConsumingEnumerable())
            {
                Console.WriteLine("Consumed " + n);
                Thread.Sleep(200);
            }

            _empty.Set();
        }

        private static void Main()
        {
            new Program().run();
        }

        private BlockingCollection<int> _queue = new BlockingCollection<int>();

        private ManualResetEvent _empty = new ManualResetEvent(false);
    }
}