TPL BufferBlock Consume Method Not Being Called

Question

I want to implement the consumer/producer pattern using the BufferBlock that runs continuously similar to the question here and the code here.

I tried to use an ActionBlock like the OP, but if the bufferblock is full and new messages are in it's queue then the new messages never get added to the ConcurrentDictionary _queue.

In the code below the ConsumeAsync method never gets called when a new message is added to the bufferblock with this call:_messageBufferBlock.SendAsync(message)

How can I correct the code below so that the ConsumeAsync method is called every time a new message is added using _messageBufferBlock.SendAsync(message)?

    public class PriorityMessageQueue 
    {
        private volatile ConcurrentDictionary<int,MyMessage> _queue = new ConcurrentDictionary<int,MyMessage>();
        private volatile BufferBlock<MyMessage> _messageBufferBlock;
        private readonly Task<bool> _initializingTask; // not used but allows for calling async method from constructor
        private int _dictionaryKey;

        public PriorityMessageQueue()
        {
            _initializingTask = Init();
        }

        public async Task<bool> EnqueueAsync(MyMessage message)
        {
            return await _messageBufferBlock.SendAsync(message);
        }

        private async Task<bool> ConsumeAsync()
        {
            try
            {
                // This code does not fire when a new message is added to the buffereblock
                while (await _messageBufferBlock.OutputAvailableAsync())
                {
                    // A message object is never received from the bufferblock
                    var message = await _messageBufferBlock.ReceiveAsync();


                }

                return true;
            }
            catch (Exception ex)
            {
                return false;
            }
        }

        private async Task<bool> Init()
        {
            var executionDataflowBlockOptions = new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = Environment.ProcessorCount,
                BoundedCapacity = 50
            };

            var prioritizeMessageBlock = new ActionBlock<MyMessage>(msg =>
            {
                SetMessagePriority(msg);
            }, executionDataflowBlockOptions);

            _messageBufferBlock = new BufferBlock<MyMessage>();
            _messageBufferBlock.LinkTo(prioritizeMessageBlock, new DataflowLinkOptions { PropagateCompletion = true, MaxMessages = 50});

            return await ConsumeAsync();
        }
    }

EDIT I have removed all the extra code and added comments.

Answer 1

I'm still not completely certain what you're trying to accomplish but I'll try to point you in the right direction. Most of the code in the example isn't strictly necessary.

I need to know when a new message arrives

If this is your only requirement then I'll assume you just need to run some arbitrary code whenever a new message is passed in. The easiest way to do that in dataflow is to use a TransformBlock and set that block as the initial receiver in your pipeline. Each block has it's own buffer so unless you have need for another buffer you can leave it out.

public class PriorityMessageQueue {        
    private TransformBlock<MyMessage, MyMessage> _messageReciever;

    public PriorityMessageQueue() {
        var executionDataflowBlockOptions = new ExecutionDataflowBlockOptions {
            MaxDegreeOfParallelism = Environment.ProcessorCount,
            BoundedCapacity = 50
        };

        var prioritizeMessageBlock = new ActionBlock<MyMessage>(msg => {
            SetMessagePriority(msg);
        }, executionDataflowBlockOptions);

        _messageReciever = new TransformBlock<MyMessage, MyMessage>(msg => NewMessageRecieved(msg), executionDataflowBlockOptions);
        _messageReciever.LinkTo(prioritizeMessageBlock, new DataflowLinkOptions { PropagateCompletion = true });
    }

    public async Task<bool> EnqueueAsync(MyMessage message) {
        return await _messageReciever.SendAsync(message);
    }

    private MyMessage NewMessageRecieved(MyMessage message) {
        //do something when a new message arrives

        //pass the message along in the pipeline
        return message;
    }

    private void SetMessagePriority(MyMessage message) {
        //Handle a message
    }
}

Of course the other option you have would be to do whatever it is you need to immediately within EnqueAsync before returning the task from SendAsync but the TransformBlock gives you extra flexibility.