Last offset fix (#407)

* Currently, the last consumed offset is stored before the message handler is called. This change does it after the messagehandler is called in order to avoid potential messageloss on the event of connection-issues/reconnecting-consumer

* Aligned the fix for consumers with superstream consumers
---------
Co-authored-by: Falk Jonas - HK <[email protected]>

Author: jonnepmyra

RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs

@@ -75,13 +75,14 @@ await OnEntityClosed(_consumerConfig.StreamSystem, _consumerConfig.Stream,
             },
             MessageHandler = async (consumer, ctx, message) =>
             {
-                _consumedFirstTime = true;
-                _lastOffsetConsumed[_consumerConfig.Stream] = ctx.Offset;
                 if (_consumerConfig.MessageHandler != null)
                 {
                     await _consumerConfig.MessageHandler(_consumerConfig.Stream, consumer, ctx, message)
                         .ConfigureAwait(false);
                 }
+
+                _consumedFirstTime = true;
+                _lastOffsetConsumed[_consumerConfig.Stream] = ctx.Offset;
             },
         }, BaseLogger).ConfigureAwait(false);
     }
@@ -146,13 +147,14 @@ await OnEntityClosed(_consumerConfig.StreamSystem, update.Stream, r,
                     },
                     MessageHandler = async (partitionStream, consumer, ctx, message) =>
                     {
-                        _consumedFirstTime = true;
-                        _lastOffsetConsumed[_consumerConfig.Stream] = ctx.Offset;
                         if (_consumerConfig.MessageHandler != null)
                         {
                             await _consumerConfig.MessageHandler(partitionStream, consumer, ctx,
                                 message).ConfigureAwait(false);
                         }
+
+                        _consumedFirstTime = true;
+                        _lastOffsetConsumed[_consumerConfig.Stream] = ctx.Offset;
                     },
                 }, BaseLogger).ConfigureAwait(false);
         }