Added better handling of ignored events

Author: alexeyzimarev

src/Eventuous.Shovel/ExtendedShovelService.cs

@@ -16,9 +16,9 @@ public class ShovelService<TSubscription, TProducer, TProduceOptions> : IHostedS
         readonly TSubscription _subscription;
         readonly TProducer     _producer;
 
-        public record ShovelMessage(string TargetStream, object Message, TProduceOptions ProduceOptions);
+        public record ShovelMessage(string TargetStream, object? Message, TProduceOptions ProduceOptions);
 
-        public delegate ValueTask<ShovelMessage> RouteAndTransform(object message);
+        public delegate ValueTask<ShovelMessage?> RouteAndTransform(object message);
 
         public delegate TSubscription CreateSubscription(
             string                     subscriptionId,
@@ -96,8 +96,15 @@ ShovelService<TSubscription, TProducer, TProduceOptions>.RouteAndTransform trans
         }
 
         public async Task HandleEvent(object evt, long? position, CancellationToken cancellationToken) {
-            var (targetStream, message, options) = await _transform(evt).NoContext();
-            await _eventProducer.Produce(targetStream, new[] { message }, options, cancellationToken).NoContext();
+            var shovelMessage = await _transform(evt).NoContext();
+            if (shovelMessage?.Message == null) return;
+
+            await _eventProducer.Produce(
+                shovelMessage.TargetStream,
+                new[] { shovelMessage.Message },
+                shovelMessage.ProduceOptions,
+                cancellationToken
+            ).NoContext();
         }
     }
 }

src/Eventuous.Shovel/ShovelService.cs

@@ -25,7 +25,7 @@ public class ShovelService<TSubscription, TProducer> : IHostedService
 
         public record ShovelMessage(string TargetStream, object? Message);
 
-        public delegate ValueTask<ShovelMessage> RouteAndTransform(object message);
+        public delegate ValueTask<ShovelMessage?> RouteAndTransform(object message);
 
         public delegate TSubscription CreateSubscription(
             string                     subscriptionId,
@@ -145,9 +145,12 @@ ShovelService<TSubscription, TProducer>.RouteAndTransform transform
         }
 
         public async Task HandleEvent(object evt, long? position, CancellationToken cancellationToken) {
-            var (targetStream, message) = await _transform(evt).NoContext();
-            if (message == null) return;
-            await _eventProducer.Produce(targetStream, new[] { message }, cancellationToken).NoContext();
+            var shovelMessage = await _transform(evt).NoContext();
+            if (shovelMessage?.Message == null) return;
+
+            await _eventProducer
+                .Produce(shovelMessage.TargetStream, new[] { shovelMessage.Message }, cancellationToken)
+                .NoContext();
         }
     }
 }