...
Next, implement your batch processing method. It is important to note that theerror message handling behavior within batch processing differs from the regular ProcessMessageAsync method.
You must
...
update each individual message context within a batch. You must mark each message context as:
(a) Errored: messageContext.WriteEvent(EventSeverity.Error, "my error");
(b) Filtered: messageContext.FilterMessage(...);
(c) Processed: Set the messageContext.BatchProcessingSuccess = true after you message has successfully processed.
| Code Block | ||
|---|---|---|
| ||
public async Task ProcessMessagesAsync(IMessageBatchContext context, CancellationToken token)
{
// iterate through each message context and process the message. Remember that
// you should be handling exceptions within your loop and setting each message's
// state. If you throw an exception out of this method, then the OnBatchError method
// will be called.
await context.Contexts.ForEachAsync(async messageContext =>
{
if(token.IsCancellationRequested)
return;
try
{
// your business logic here...
await Task.Delay(500, token).ConfigureAwait(false);
// ** you *must* mark your context as successfully processed **
messageContext.BatchProcessingSuccess = true;
}
catch(OperationCanceledException)
{
return;
}
catch(Exception ex)
{
// message moved to the error queue when you add an error event
messageContext.WriteEvent(EventSeverity.Error, ex);
}
}, 10).ConfigureAwait(false);
} |