...
| 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 **
context.MarkSuccessfullyProcessedForThisDevice(messageContext);
}
catch(OperationCanceledException)
{
// exit when the device/channel is stopped
return;
}
catch(Exception ex)
{
// message moved to the error queue when you add an error event
messageContext.WriteEvent(EventSeverity.Error, ex);
}
}, 10).ConfigureAwait(false);
} |
| Code Block |
|---|
|
// Called when an exception is thrown from the ProcessMessagesAsync method.
// *This only applies when you are processing the entire batch in one go. If you
// *are processing messages within the batch individually (or in smaller batches,
// *you need to handle exceptions within the ProcessMessagesAsync method.
public override void OnBatchError(IMessageBatchContext context, Connexion.Core.ErrorEventArgs args)
{
if(args.Exception is OperationCanceledException)
return; // stopped device/channel
args.ShouldRetry = true;
args.SleepTime = TimeSpan.FromMinutes(2);
// this error is displayed over your device
Logger.Write(EventSeverity.Error, 54321, args.Exception.Message);
}
Note that in the example above, messages are being processed in parallel. In this scenario, it's possible for some messages to succeed and some messages to fail. Individual messages that do not succeed will either be moved to the error queue, or, retried on a subsequent attempt. Another batch processing case involves the entire batch of messages succeeding or failing as a single batch. An example of this would be the sending of a batch of messages to an external service. In this case, you may serialize all the messages in the batch into a single payload, and send that payload. If the payload succeeds, all messages should be successful, and if it does not, the entire batch should be retried.
In this scenario, the context.MarkSuccessfullyProcessedForThisDevice(messageContext) has no effect, and is not required. Messages will be marked as processed once the final device has been reached (without errors).
Please note - there is no built-in retry or error handling policy - it is up to you to implement retry logic and error handling. Here is an example showing a retry policy: