I'm trying to create a lifetime process that batches incoming messages for DB bulk insert.
The new message is coming in 1 at a time, in an irregular interval.
My solution to this would be something like the producer-consumer pattern using BlockingCollection
.
Messages are added freely into the BlockingCollection
by various events and are taken out of BlockingCollection
in bulk for DB insert in regular intervals, 5 seconds.
However, the current solution is fire-and-forget. If the bulk insert failed for any reason, I need a way for the processor to notify the original sources of the failure, because the source contains the logic to recover and retry.
Is there a specific pattern I should be using for what I'm trying to achieve? Any suggestion or help is much appreciated!
private BlockingCollection<Message> _messageCollection;
public async Task<bool> InsertMessage(Message message)
{
if (!_messageCollection.TryAdd(message)) return false;
// TODO: check message has been successfully processed, if not return false
// return false;
return true;
}
private void BulkInsertProcess()
{
Task consumerThread = Task.Factory.StartNew(async () =>
{
while (!_messageCollection.IsCompleted)
{
List<Message> messages = new List<Message>();
for (int i = 0; i < 50; i++)
{
if (_messageCollection.Any())
{
messages.Add(_messageCollection.Take());
}
else
{
break;
}
}
bool insertResult = await _database.BulkInsertMessages(messages);
// TODO: check result and inform the consumer if insert failed
await Task.Delay(5000);
}
});
}
TaskCompletionSource
with my message class, I am able toSetResult()
orSetException()
at the consumer task, which the source can await on or try catch any exception.