...
(c) Processed: Set the messageContext.BatchProcessingSuccess = true after you message has successfully processed.
The following example loops through each message context within the batch. If the cancellation token has been set (by stopping the channel or pausing the device), then any non-processed messages are returned to the queued state. Any messages which have been successfully processed (or errored/filtered) will be filed into the associated queue.
| 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);
} |
You should not throw exceptions out of the BatchProcessMessages method, however, if you do, there is an OnBatchError method which will be called. Within this method you can choose to retry the entire batch, or, error any messages in the batch which are not processed/errored/filtered. If you are sending the entire batch of messages to an external service, then you can throw an exception from ProcessMessagesAsync and set the ShouldRetry property in the OnBatchError method to true. This will cause the entire batch to be retried.
| Code Block | ||
|---|---|---|
| ||
// Called when an exception is thrown from the ProcessMessagesAsync method
public override void OnBatchError(IMessageBatchContext context, Connexion.Core.ErrorEventArgs args)
{
if(args.Exception is OperationCanceledException)
return;
args.ShouldRetry = true;
args.SleepTime = TimeSpan.FromMinutes(2);
// this error is displayed over your device
Logger.Write(EventSeverity.Error, 54321, args.Exception.Message);
} |