Avoid resubscribe from the event, which is the next to the failed one

alexeyzimarev · alexeyzimarev · commit 9f4cb7aebb7f · 2021-06-15T12:43:32.000+02:00
diff --git a/src/Eventuous.Subscriptions/SubscriptionService.cs b/src/Eventuous.Subscriptions/SubscriptionService.cs
@@ -75,9 +75,8 @@ protected async Task Handler(ReceivedEvent re, CancellationToken cancellationTok
                 re.EventType
             );
 
-            _lastProcessed = GetPosition(re);
-
             if (re.EventType.StartsWith("$")) {
+                _lastProcessed = GetPosition(re);
                 await Store().NoContext();
                 return;
             }
@@ -88,6 +87,7 @@ await Task.WhenAll(
                         _eventHandlers.Select(x => x.HandleEvent(re.Payload, (long?) re.StreamPosition, cancellationToken))
                     ).NoContext();
                 }
+                _lastProcessed = GetPosition(re);
             }
             catch (Exception e) {
                 Log?.Log(

Original file line number	Diff line number	Diff line change
`@@ -75,9 +75,8 @@ protected async Task Handler(ReceivedEvent re, CancellationToken cancellationTok`
`75`	`75`	`re.EventType`
`76`	`76`	`);`
`77`	`77`
`78`		`- _lastProcessed = GetPosition(re);`
`79`		`-`
`80`	`78`	`if (re.EventType.StartsWith("$")) {`
	`79`	`+ _lastProcessed = GetPosition(re);`
`81`	`80`	`await Store().NoContext();`
`82`	`81`	`return;`
`83`	`82`	`}`
`@@ -88,6 +87,7 @@ await Task.WhenAll(`
`88`	`87`	`_eventHandlers.Select(x => x.HandleEvent(re.Payload, (long?) re.StreamPosition, cancellationToken))`
`89`	`88`	`).NoContext();`
`90`	`89`	`}`
	`90`	`+ _lastProcessed = GetPosition(re);`
`91`	`91`	`}`
`92`	`92`	`catch (Exception e) {`
`93`	`93`	`Log?.Log(`