Merge pull request #8 from Cabazure/hotfix/checkpoint

Ensure EventHubProcessor updates partition checkpoints

Author: rickykaare

src/Cabazure.Messaging.EventHub/Internal/EventHubBatchHandler.cs

@@ -9,7 +9,7 @@ public interface IEventHubBatchHandler<TMessage, TProcessor>
 {
     public TProcessor Processor { get; }
 
-    Task ProcessBatchAsync(
+    Task<EventData?> ProcessBatchAsync(
         IEnumerable<EventData> events,
         EventProcessorPartition partition,
         CancellationToken cancellationToken);
@@ -29,29 +29,47 @@ public class EventHubBatchHandler<TMessage, TProcessor>(
 {
     public TProcessor Processor => processor;
 
-    public async Task ProcessBatchAsync(
+    public async Task<EventData?> ProcessBatchAsync(
        IEnumerable<EventData> events,
        EventProcessorPartition partition,
        CancellationToken cancellationToken)
     {
+        EventData? lastEvent = null;
         foreach (var evt in events)
         {
-            if (!filters.TrueForAll(f => f.Invoke(evt.Properties)))
+            lastEvent = evt;
+
+            try
             {
-                continue;
-            }
+                if (!filters.TrueForAll(f => f.Invoke(evt.Properties)))
+                {
+                    continue;
+                }
 
-            var message = evt.EventBody
-                .ToObjectFromJson<TMessage>(serializerOptions);
+                var message = evt.EventBody
+                    .ToObjectFromJson<TMessage>(serializerOptions);
 
-            var metadata = EventHubMetadata
-                .Create(evt, partition.PartitionId);
+                if (message != null)
+                {
+                    var metadata = EventHubMetadata
+                        .Create(evt, partition.PartitionId);
 
-            await processor.ProcessAsync(
-                message!,
-                metadata,
-                cancellationToken);
+                    await processor
+                        .ProcessAsync(
+                            message!,
+                            metadata,
+                            cancellationToken)
+                        .ConfigureAwait(false);
+                }
+            }
+            catch (Exception ex)
+            {
+                await ProcessErrorAsync(ex, cancellationToken)
+                    .ConfigureAwait(false);
+            }
         }
+
+        return lastEvent;
     }
 
     public async Task ProcessErrorAsync(

src/Cabazure.Messaging.EventHub/Internal/EventHubProcessor.cs

@@ -1,6 +1,7 @@
 using Azure.Core;
 using Azure.Messaging.EventHubs;
 using Azure.Messaging.EventHubs.Primitives;
+using Azure.Messaging.EventHubs.Processor;
 
 namespace Cabazure.Messaging.EventHub.Internal;
 
@@ -53,13 +54,24 @@ protected override async Task OnProcessingEventBatchAsync(
         IEnumerable<EventData> events,
         EventProcessorPartition partition,
         CancellationToken cancellationToken)
-        => await batchHandler
+    {
+        var lastEvent = await batchHandler
             .ProcessBatchAsync(
                 events,
                 partition,
                 cancellationToken)
             .ConfigureAwait(false);
 
+        if (lastEvent != null)
+        {
+            await UpdateCheckpointAsync(
+                    partition.PartitionId,
+                    CheckpointPosition.FromEvent(lastEvent),
+                    cancellationToken)
+                .ConfigureAwait(false);
+        }
+    }
+
     protected override async Task OnProcessingErrorAsync(
         Exception exception,
         EventProcessorPartition partition,