Expose Store offset to stream system (#408)

* Add the ability to store offsets using the streamsystem, which can be used when you're not in the context of a messagehandler

* Added reconnection and a test

---------

Co-authored-by: Falk Jonas - HK <[email protected]>

Author: Gsantomaggio

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -338,6 +338,7 @@ RabbitMQ.Stream.Client.StreamSystem.CreateRawSuperStreamProducer(RabbitMQ.Stream
 RabbitMQ.Stream.Client.StreamSystem.CreateSuperStream(RabbitMQ.Stream.Client.SuperStreamSpec spec) -> System.Threading.Tasks.Task
 RabbitMQ.Stream.Client.StreamSystem.CreateSuperStreamConsumer(RabbitMQ.Stream.Client.RawSuperStreamConsumerConfig rawSuperStreamConsumerConfig, Microsoft.Extensions.Logging.ILogger logger = null) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ISuperStreamConsumer>
 RabbitMQ.Stream.Client.StreamSystem.DeleteSuperStream(string superStream) -> System.Threading.Tasks.Task
+RabbitMQ.Stream.Client.StreamSystem.StoreOffset(string reference, string stream, ulong offsetValue) -> System.Threading.Tasks.Task
 RabbitMQ.Stream.Client.StreamSystem.StreamInfo(string streamName) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.StreamInfo>
 RabbitMQ.Stream.Client.StreamSystem.StreamStats(string stream) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.StreamStats>
 RabbitMQ.Stream.Client.StreamSystem.SuperStreamExists(string superStream) -> System.Threading.Tasks.Task<bool>

RabbitMQ.Stream.Client/StreamSystem.cs

@@ -184,6 +184,12 @@ private async Task MayBeReconnectLocator()
             }
         }
 
+        public async Task StoreOffset(string reference, string stream, ulong offsetValue)
+        {
+            await MayBeReconnectLocator().ConfigureAwait(false);
+            await _client.StoreOffset(reference, stream, offsetValue).ConfigureAwait(false);
+        }
+
         public async Task UpdateSecret(string newSecret)
         {
             // store the old password just in case it will fail to update the secret

Tests/SystemTests.cs

@@ -401,5 +401,18 @@ await Assert.ThrowsAsync<DeleteStreamException>(
             );
             await system.Close();
         }
+
+        [Fact]
+        public async Task ClientShouldStoreOffset()
+        {
+            var stream = Guid.NewGuid().ToString();
+            var consumerRef = "myRef";
+            var system = await StreamSystem.Create(new StreamSystemConfig());
+            await system.CreateStream(new StreamSpec(stream));
+            await system.StoreOffset(consumerRef, stream, 4);
+            Assert.Equal((ulong)4, await system.QueryOffset(consumerRef, stream));
+            await system.DeleteStream(stream);
+            await system.Close();
+        }
     }
 }