Asynchronous Take from blocking collection

Viewed 7331

I'm using a BlockingCollection to implement a producer/consumer pattern. I have an asynchronous loop that fills the collection with data to be processed which can then at a much later time be accessed by the client. The packets arrive sparsely and I would like the polling to be done without using a blocking call to take.

In essence, I'm looking for something like a BeginTake and EndTake that does not exist in the blocking collection so that I can make use of the internal thread pool in a callback. It does not have to be a BlockingCollection by any means. Anything that does what I need would be great.

This is what I've got now. _bufferedPackets is a BlockingCollection<byte[]>:

public byte[] Read(int timeout)
{
    byte[] result;
    if (_bufferedPackets.IsCompleted)
    {
        throw new Exception("Out of packets");
    }
    _bufferedPackets.TryTake(out result, timeout);      
    return result;
}

I'd like this to be something like this, in pseudocode:

public void Read(int timeout)
{
    _bufferedPackets.BeginTake(result =>
        {
            var bytes = _bufferedPackets.EndTake(result);
            // Process the bytes, or the resuting timeout
        }, timeout, _bufferedPackets);
}

What are my options for this? I do not want to place any thread in a waiting state, since there is lots of other IO stuff for it to process, and I would run out of threads pretty quickly.

Update: I've rewritten the code in question to use the async process differently, essentially swapping callbacks based on if there is a waiting request within the timeout limit. This works fine, but still it would be awesome if there was a way to do this without resorting to timers and swapping lambdas around which potentially causes race conditions and is hard to write (and understand). I have solved this also with an own implementation of an async queue, but it would still be awesome if there was a more standard and well tested option.

3 Answers
Related