BrokeredMessage is automatically removed after calling OnMessage ()

It's incredibly easy to do with BlockingCollection

.

var batchingQueue = new BlockingCollection<BrokeredMessage>();

myQueueClient.OnMessage((m) =>
{
    Console.WriteLine("Queueing message");
    batchingQueue.Add(m);
});

      

        
        
        
      

    

And your consumer flow:

foreach (var msg in batchingQueue.GetConsumingEnumerable())
{
    Console.WriteLine("Completing message");
    msg.Complete();
}

      

        
        
        
      

    

GetConsumingEnumerable returns an iterator that consumes items in the queue until the property is set IsCompleted

and the queue is empty. If the queue is empty but IsCompleted

is False

, it waits for an unoccupied wait for the next item.

To cancel the consumer thread (i.e. shutdown the program), you will stop adding things to the queue and get a call to the main thread batchingQueue.CompleteAdding

. The consumer will empty the queue, see that the property IsCompleted

has a value, True

and exit.

The usage BlockingCollection

here is better than ConcurrentBag

or ConcurrentQueue

because the interface is BlockingCollection

easier to work with. In particular, usage GetConsumingEnumerable

saves you from having to worry about checking an account or busy waiting (polling loops). It just works.

Also note that it ConcurrentBag

has a rather strange behavior when deleting. In particular, the order of deleting elements differs depending on which thread is deleting the element. The thread that created the bag removes the items in a different order than the other threads. See Using the ConcurrentBag Collection for details .

You have not specified why you want to inject items in the input. Unless there are fundamental reasons for creating performance, it is not a good idea to complicate your code with this dispensing logic.

If you want to do a batch write to the database, I would suggest using a simple one List<T>

to buffer items. If you need to process items before writing them to the database, use the technique shown above to process them. Then, instead of writing directly to the database, add the item to the list. When the list receives 1000 items or the specified amount of time has elapsed, select a new list and run a task to write the old list to the database. Like this:

// at class scope

// Flush every 5 minutes.
private readonly TimeSpan FlushDelay = TimeSpan.FromMinutes(5);
private const int MaxBufferItems = 1000;

// Create a timer for the buffer flush.
System.Threading.Timer _flushTimer = new System.Threading.Timer(TimedFlush, FlushDelay.TotalMilliseconds, Timeout.Infinite);  

// A lock for the list. Unless you're getting hundreds of thousands
// of items per second, this will not be a performance problem.
object _listLock = new Object();

List<BrokeredMessage> _recordBuffer = new List<BrokeredMessage>();

      

        
        
        
      

    

Then your consumer:

foreach (var msg in batchingQueue.GetConsumingEnumerable())
{
    // process the message
    Console.WriteLine("Completing message");
    msg.Complete();
    lock (_listLock)
    {
        _recordBuffer.Add(msg);
        if (_recordBuffer.Count >= MaxBufferItems)
        {
            // Stop the timer
            _flushTimer.Change(Timeout.Infinite, Timeout.Infinite);

            // Save the old list and allocate a new one
            var myList = _recordBuffer;
            _recordBuffer = new List<BrokeredMessage>();

            // Start a task to write to the database
            Task.Factory.StartNew(() => FlushBuffer(myList));

            // Restart the timer
            _flushTimer.Change(FlushDelay.TotalMilliseconds, Timeout.Infinite);
        }
    }
}

private void TimedFlush()
{
    bool lockTaken = false;
    List<BrokeredMessage> myList = null;

    try
    {
        if (Monitor.TryEnter(_listLock, 0, out lockTaken))
        {
            // Save the old list and allocate a new one
            myList = _recordBuffer;
            _recordBuffer = new List<BrokeredMessage>();
        }
    }
    finally
    {
        if (lockTaken)
        {
            Monitor.Exit(_listLock);
        }
    }

    if (myList != null)
    {
        FlushBuffer(myList);
    }

    // Restart the timer
    _flushTimer.Change(FlushDelay.TotalMilliseconds, Timeout.Infinite);
}

      

        
        
        
      

    

The idea here is that you delete the old list, highlight the new list to continue processing, and then write the old list items to the database. The interlock prevents the timer and the counter of recordings from being activated on each other. Without blocking, everything will most likely work fine for a while, and then you get weird crashes at unpredictable times.

I like this design because it excludes consumer polling. The only thing I don't like is that the consumer needs to know the timer (i.e. it should stop and then restart the timer). With a little thought, I could eliminate this requirement. But he works well as he wrote.

I had the same problem when I started working with Azure Service Bus.

I found that the OnMessage method always deletes the BrokedMessage object. The approach suggested by Jim Michel did not work for me (but it was very interesting to read - thanks!).

After some investigation, I found that the whole approach was flawed. Let me explain the correct way to do what you want.

• Use the BrokedMessage.Complete () method only inside the OnMessage method handler.
• If you need to process the message outside of this method, you should use the QueueClient.Complete (Guid lockToken) method. "LockToken" is a property of the BrokeredMessage object.

Example:

 var messageOptions = new OnMessageOptions {
      AutoComplete       = false,
      AutoRenewTimeout   = TimeSpan.FromMinutes( 5 ),
     MaxConcurrentCalls = 1
 };
 var buffer = new Dictionary<string, Guid>();

 // get message from queue 
 myQueueClient.OnMessage(
      m => buffer.Add(key: m.GetBody<string>(), value: m.LockToken), 
      messageOptions // this option says to ServiceBus to "froze" message in he queue until we process it
 );         

 foreach(var item in buffer){
    try {
        Console.WriteLine($"Process item: {item.Key}");
        myQueueClient.Complete(item.Value);// you can also use method CompleteBatch(...) to improve performance
    } 
    catch{
        // "unfroze" message in ServiceBus. Message would be delivered to other listener 
        myQueueClient.Defer(item.Value);
    }
 }