As part of a much larger solution, I have a solution for the producer-consumer pattern in .NET. It uses a System.Threading.Channels.Channel for receiving messages from a WebSocket and allowing it to be processed as the messages arrive.

using System.Threading.Channels;

/// <summary>
/// A class that asynchronously dispatches items in a first-in-first-out manner,
/// using a separate thread for dispatching the items.
/// </summary>
/// <typeparam name="T">The type of items to be dispatched.</typeparam>
public class Dispatcher<T>
{
    private readonly Channel<T> queue = Channel.CreateUnbounded<T>(new UnboundedChannelOptions()
    {
        SingleReader = true,
        SingleWriter = true,
    });

    private readonly Task monitorTask;
    private bool isDispatching;

    /// <summary>
    /// Initializes a new instance of the <see cref="Dispatcher{T}"/> class.
    /// </summary>
    public Dispatcher()
    {
        this.monitorTask = Task.Run(async () => await this.MonitorQueue().ConfigureAwait(false));
        this.isDispatching = true;
    }

    /// <summary>
    /// Occurs when an item is dispatched.
    /// </summary>
    public event EventHandler<ItemDispatchedEventArgs<T>>? ItemDispatched;

    /// <summary>
    /// Gets a value indicating whether the dispatcher is currently dispatching items.
    /// </summary>
    public bool IsDispatching => this.isDispatching;

    /// <summary>
    /// Attempts to add an item to the queue to be dispatched.
    /// </summary>
    /// <param name="itemToDispatch">The item to be dispatched.</param>
    /// <returns><see langword="true"/> if the item was queued for dispatching; otherwise, <see langword="false"/>.</returns>
    public bool TryDispatch(T itemToDispatch)
    {
        return this.queue.Writer.TryWrite(itemToDispatch);
    }

    /// <summary>
    /// Shuts down the dispatcher.
    /// </summary>
    public void StopDispatching()
    {
        if (this.isDispatching)
        {
            // Attempt to wait for the channel to empty before marking the
            // writer as complete and waiting for the monitor task to end.
            while (this.queue.Reader.TryPeek(out _))
            {
                // N.B. We are doing an explicit .Wait() call here to avoid
                // having this become an async method. Since all we are doing
                // is waiting for the reader to become empty, this should be
                // an acceptable use of a synchronizing API on a typically
                // async structure. If this becomes an issue on shutdown
                // of the Dispatcher, we can add a configurable shutdown timeout.
                Task.Delay(TimeSpan.FromMilliseconds(10)).Wait();
            }

            this.queue.Writer.Complete();
            this.monitorTask.Wait();
            this.isDispatching = false;
        }
    }

    /// <summary>
    /// Raises the ItemDispatched event.
    /// </summary>
    /// <param name="sender">The object raising the event.</param>
    /// <param name="e">The EventArgs containing information about the event.</param>
    protected virtual void OnItemDispatched(object? sender, ItemDispatchedEventArgs<T> e)
    {
        if (this.ItemDispatched is not null)
        {
            this.ItemDispatched(this, e);
        }
    }

    private async Task MonitorQueue()
    {
        while (await this.queue.Reader.WaitToReadAsync().ConfigureAwait(false))
        {
            this.DispatchPendingItems();
        }
    }

    private void DispatchPendingItems()
    {
        while (this.queue.Reader.TryRead(out T? item))
        {
            this.OnItemDispatched(this, new ItemDispatchedEventArgs<T>(item));
        }
   }
}

This works remarkably well, but suppose I wanted to replace this implementation with one based on System.Collections.Concurrent.BlockingCollection. My initial pass at such an implementation looks like this:

using System.Collections.Concurrent;

/// <summary>
/// A class that asynchronously dispatches items in a first-in-first-out manner,
/// using a separate thread for dispatching the items.
/// </summary>
/// <typeparam name="T">The type of items to be dispatched.</typeparam>
public class Dispatcher<T>
{
    private readonly BlockingCollection<T> items = new();
    private readonly CancellationTokenSource monitorCancellationTokenSource = new();
    private readonly Task monitorTask;
    private bool isDispatching;

    /// <summary>
    /// Initializes a new instance of the <see cref="Dispatcher{T}"/> class.
    /// </summary>
    public Dispatcher()
    {
        this.monitorTask = Task.Run(() => this.MonitorQueue());
        this.isDispatching = true;
    }

    /// <summary>
    /// Occurs when an item is dispatched.
    /// </summary>
    public event EventHandler<ItemDispatchedEventArgs<T>>? ItemDispatched;

    /// <summary>
    /// Gets a value indicating whether the dispatcher is currently dispatching items.
    /// </summary>
    public bool IsDispatching => this.isDispatching;

    /// <summary>
    /// Attempts to add an item to the queue to be dispatched.
    /// </summary>
    /// <param name="itemToDispatch">The item to be dispatched.</param>
    /// <returns><see langword="true"/> if the item was queued for dispatching; otherwise, <see langword="false"/>.</returns>
    public bool TryDispatch(T itemToDispatch)
    {
        if (this.items.IsAddingCompleted)
        {
            return false;
        }

        this.items.Add(itemToDispatch);
        return true;
    }

    /// <summary>
    /// Shuts down the dispatcher.
    /// </summary>
    public void StopDispatching()
    {
        if (this.isDispatching)
        {
            // Attempt to wait for the channel to empty before marking the
            // writer as complete and waiting for the monitor task to end.
            while (this.items.Count > 0)
            {
                // N.B. We are doing an explicit .Wait() call here to avoid
                // having this become an async method. Since all we are doing
                // is waiting for the reader to become empty, this should be
                // an acceptable use of a synchronizing API on a typically
                // async structure. If this becomes an issue on shutdown
                // of the Dispatcher, we can add a configurable shutdown timeout.
                Task.Delay(TimeSpan.FromMilliseconds(10)).Wait();
            }

            this.items.CompleteAdding();
            this.monitorCancellationTokenSource.Cancel();
            this.monitorTask.Wait();
            this.isDispatching = false;
        }
    }

    /// <summary>
    /// Raises the ItemDispatched event.
    /// </summary>
    /// <param name="sender">The object raising the event.</param>
    /// <param name="e">The EventArgs containing information about the event.</param>
    protected virtual void OnItemDispatched(object? sender, ItemDispatchedEventArgs<T> e)
    {
        if (this.ItemDispatched is not null)
        {
            this.ItemDispatched(this, e);
        }
    }

    private void MonitorQueue()
    {
        try
        {
            foreach (T itemToDispatch in this.items.GetConsumingEnumerable(this.monitorCancellationTokenSource.Token))
            {
                this.OnItemDispatched(this, new ItemDispatchedEventArgs<T>(itemToDispatch));
            }
        }
        catch (OperationCanceledException)
        {
        }
    }
}

Now to the root of my question. In my unit tests for the larger project, I have a large body of tests that do the equivalent of:

[Test]
public void TestSomething
{
    // Unit tests don't actually create an independent Dispatcher
    // instance, but does so in context of other objects with
    // proper event handlers hooked up.
    Dispatcher dispatcher = new Dispatcher<string>();
    dispatcher.OnItemDispatched += (sender, e) => Console.WriteLine(e.DispatchedItem);

    // Do some actions to get items dispatched
    // N.B., tests never explicitly call dispatcher.StopDispatching().
}

When I use the Channel-based implementation, I can run many, many unit tests consecutively. I’ve not yet encountered a limit to what I can do. When I use the BlockingCollection-based implementation, I can only run a certain number of unit tests before things start going awry. I get items not being consumed by the consumer (not being dispatched). Here’s the question: Why?

I have my suspicions on why this would be the case, but I’d love to get some wisdom from other, more knowledgeable people than I. If one of my implementations is incorrect, I’d like to know how to correct it. If it’s down to the internals of the .NET classes, I’d like to know that too. Yes, I’m aware that I should clean up the usage in the tests to dispose of the objects properly, but that is orthogonal to the question I’m asking: Why does it work in one case and not the other?