State

Author: Rafał Maciąg

src/MicroPlumberd.ProcessManager.Abstractions/IProcessManager.cs

@@ -22,6 +22,7 @@ public abstract class ProcessManagerBase<TId> : IVersionAware, IIdAware, IId<TId
         object IIdAware.Id { set => _id = (TId)value; }
         public long Version => _version;
         void IVersionAware.Increase() => _version += 1;
+        void IVersionAware.SetValue(long nv) { _version = nv; }
 
         public virtual async Task<ICommandRequest?> HandleError(ExecutionContext executionContext) => null;
     }

src/MicroPlumberd.Services.ProcessManager/ProcessManagerClient.cs

@@ -78,8 +78,8 @@ public async Task<IAsyncDisposable> SubscribeProcessManager<TProcessManager>() w
             ProcessManagerExecutor<TProcessManager>.Sender sender =
                 new ProcessManagerExecutor<TProcessManager>.Sender(this);
             var c = AsyncDisposableCollection.New();
-            c += await Plumber.SubscribeEventHandlerPersistently(sender, $"{typeof(TProcessManager).Name}Outbox", ensureOutputStreamProjection:true);
-            c += await Plumber.SubscribeEventHandlerPersistently(executor, $"{typeof(TProcessManager).Name}Inbox", ensureOutputStreamProjection:true);
+            c += await Plumber.SubscribeEventHandlerPersistently(sender, $"{typeof(TProcessManager).Name}Outbox", ensureOutputStreamProjection: true);
+            c += await Plumber.SubscribeEventHandlerPersistently(executor, $"{typeof(TProcessManager).Name}Inbox", ensureOutputStreamProjection: true);
 
             await Plumber.ProjectionManagementClient.EnsureLookupProjection(Plumber.ProjectionRegister,
                 typeof(TProcessManager).Name, "RecipientId", $"{typeof(TProcessManager).Name}Lookup");

src/MicroPlumberd.Services/CommandHandlerStarter.cs

@@ -44,7 +44,7 @@ public static StreamPosition ToStreamPosition(this FromStream fs)
 class EventHandlerStarter<THandler>(IPlumber plumber) : IEventHandlerStarter
     where THandler : class, IEventHandler, ITypeRegister
 {
-    public async Task Start()
+    public async Task Start(CancellationToken stoppingToken)
     {
         if (!Persistently)
             await plumber.SubscribeEventHandler<THandler>(start: StartPosition);

src/MicroPlumberd.Services/EventHandlerService.cs

@@ -11,7 +11,7 @@ sealed class EventHandlerService(IEnumerable<IEventHandlerStarter> starters) : B
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         foreach (var i in starters) 
-            await i.Start();
+            await i.Start(stoppingToken);
     }
 
 }

src/MicroPlumberd.Services/ICommandHandlerStarter.cs

@@ -2,7 +2,7 @@
 
 interface IEventHandlerStarter
 {
-    Task Start();
+    Task Start(CancellationToken stoppingToken);
 }
 interface ICommandHandlerStarter
 {

src/MicroPlumberd.Tests.App/FooDomain/FooAggregate.cs

@@ -7,7 +7,13 @@
 
 namespace MicroPlumberd.Tests.App.Domain;
 
-
+public record FooEntityState : IId<Guid>, IVersionAware
+{
+    public Guid Id { get; set; } = Guid.NewGuid();
+    public string Name { get; set; }
+    public long Version { get; private set; } = -1;
+    void IVersionAware.SetValue(long nv) => Version = nv;
+}
 
 [Aggregate]
 [ThrowsFaultException<BusinessFaultException>()]

src/MicroPlumberd.Tests/Integration/AggregateTests.cs

@@ -23,6 +23,51 @@ public void GivenMethodsAreAccessible()
         }
     }
 
+    [TestCategory("Integration")]
+    public class StateTests : IClassFixture<EventStoreServer>
+    {
+        private readonly IPlumber plumber;
+        private readonly EventStoreServer es;
+
+        public StateTests(EventStoreServer es)
+        {
+            plumber = Plumber.Create(es.GetEventStoreSettings());
+            this.es = es;
+        }
+
+        [Fact]
+        public async Task DoubleWrite()
+        {
+            await es.StartInDocker();
+
+            var st = new FooEntityState() { Name = "Foo"};
+            await plumber.AppendState(st);
+
+            st.Name = "Bar";
+            await plumber.AppendState(st);
+
+            FooEntityState actual = await plumber.GetState<FooEntityState>(st.Id);
+
+            actual.Should().Be(st);
+        }
+        [Fact]
+        public async Task WriteAfterGet()
+        {
+            await es.StartInDocker();
+
+            var st = new FooEntityState() { Name = "Foo" };
+            await plumber.AppendState(st);
+            
+            FooEntityState nx = await plumber.GetState<FooEntityState>(st.Id);
+            nx.Name = "Bar";
+            await plumber.AppendState(nx);
+
+            FooEntityState actual = await plumber.GetState<FooEntityState>(st.Id);
+            actual.Should().Be(nx);
+
+        }
+    }
+
     [TestCategory("Integration")]
     public class AggregateTests : IClassFixture<EventStoreServer>
     {

src/MicroPlumberd.Tests/Integration/ReadModelTests.cs

@@ -30,7 +30,7 @@ public async Task SubscribeModelPersistently()
 
         var fooModel = new FooModel(new InMemoryAssertionDb());
 
-        var sub = await plumber.SubscribeEventHandlerPersistently(fooModel, startFrom:StreamPosition.Start);
+        var sub = await plumber.SubscribeEventHandlerPersistently(fooModel, startFrom: StreamPosition.Start);
 
         await Task.Delay(1000);
 

src/MicroPlumberd/Abstractions/IAggregate.cs

@@ -79,7 +79,8 @@ public interface IVersioned
 /// <seealso cref="MicroPlumberd.IVersioned" />
 public interface IVersionAware : IVersioned
 {
-    void Increase();
+    void Increase() => SetValue(this.Version + 1);
+    void SetValue(long nv);
 }
 static class Extensions
 {

src/MicroPlumberd/Abstractions/IPlumber.cs

@@ -180,7 +180,7 @@ Task<IWriteResult> AppendEvents(string streamId, StreamState state, IEnumerable<
     /// <param name="userCredentials">The user credentials.</param>
     /// <param name="cancellationToken">The cancellation token.</param>
     /// <returns></returns>
-    ISubscriptionRunner Subscribe(string streamName, FromRelativeStreamPosition start, UserCredentials? userCredentials = null, CancellationToken cancellationToken = new CancellationToken());
+    ISubscriptionRunner Subscribe(string streamName, FromRelativeStreamPosition start, UserCredentials? userCredentials = null, CancellationToken cancellationToken = default);
 
     /// <summary>
     /// Subscribes the event handler. EventHandler is a class that contains many overloaded 'Given' methods. A projection will be created at EventStore that creates a joined stream from all supported event-types by EventHandler.
@@ -211,11 +211,12 @@ Task<IAsyncDisposable> SubscribeEventHandler<TEventHandler>(TypeEventConverter m
     /// <param name="outputStream">The output stream.</param>
     /// <param name="start">The start.</param>
     /// <param name="ensureOutputStreamProjection">if set to <c>true</c> [ensure output stream projection].</param>
+    /// <param name="token"></param>
     /// <returns></returns>
-    Task<IAsyncDisposable> SubscribeEventHandler<TEventHandler>(TEventHandler? eh = default, string? outputStream = null, FromRelativeStreamPosition? start = null, bool ensureOutputStreamProjection = true)
+    Task<IAsyncDisposable> SubscribeEventHandler<TEventHandler>(TEventHandler? eh = default, string? outputStream = null, FromRelativeStreamPosition? start = null, bool ensureOutputStreamProjection = true, CancellationToken token = default)
         where TEventHandler : class, IEventHandler, ITypeRegister;
 
-   
+
     /// <summary>
     /// Subscribes the event handler persistently. EventHandler is a class that contains many overloaded 'Given' methods. A projection will be created at EventStore that creates a joined stream from all supported event-types by EventHandler.
     /// Then EventHandler subscribe the the output stream.
@@ -226,8 +227,9 @@ Task<IAsyncDisposable> SubscribeEventHandler<TEventHandler>(TEventHandler? eh =
     /// <param name="groupName">Optional group name.</param>
     /// <param name="startFrom">Optional start of the stream.</param>
     /// <param name="ensureOutputStreamProjection">when true creates projection that creates output's stream</param>
+    /// <param name="token"></param>
     /// <returns></returns>
-    Task<IAsyncDisposable> SubscribeEventHandlerPersistently<TEventHandler>(TEventHandler? model=null, string? outputStream = null, string? groupName = null, IPosition? startFrom = null, bool ensureOutputStreamProjection = true) where TEventHandler : class,IEventHandler, ITypeRegister;
+    Task<IAsyncDisposable> SubscribeEventHandlerPersistently<TEventHandler>(TEventHandler? model=null, string? outputStream = null, string? groupName = null, IPosition? startFrom = null, bool ensureOutputStreamProjection = true, CancellationToken token = default) where TEventHandler : class,IEventHandler, ITypeRegister;
 
     /// <summary>
     /// Returns a subscription builder that will subscribe model persistently.
@@ -333,11 +335,13 @@ Task<IWriteResult> AppendLink(string streamId, Metadata metadata, StreamState? s
     /// <param name="groupName">Name of the group.</param>
     /// <param name="startFrom">The start from.</param>
     /// <param name="ensureOutputStreamProjection">if set to <c>true</c> [ensure output stream projection].</param>
+    /// <param name="token"></param>
     /// <returns></returns>
     Task<IAsyncDisposable> SubscribeEventHandlerPersistently<TEventHandler>(TypeEventConverter mapFunc,
         IEnumerable<string>? events,
         TEventHandler? model,
-        string? outputStream = null, string? groupName = null, IPosition? startFrom = null, bool ensureOutputStreamProjection = true)
+        string? outputStream = null, string? groupName = null, IPosition? startFrom = null,
+        bool ensureOutputStreamProjection = true, CancellationToken token = default)
         where TEventHandler : class, IEventHandler;
 
     /// <summary>

src/MicroPlumberd/EventStoreProjectionManagementClientExtensions.cs

@@ -16,42 +16,42 @@ public static async Task EnsureJoinProjection(this EventStoreProjectionManagemen
             await client.CreateContinuousAsync(outputStream, query, true);
     }
 
-    private static async Task Update(EventStoreProjectionManagementClient client, string outputStream, string query)
+    private static async Task Update(EventStoreProjectionManagementClient client, string outputStream, string query, CancellationToken token = default)
     {
-        var state = await client.GetStatusAsync(outputStream);
+        var state = await client.GetStatusAsync(outputStream, cancellationToken: token);
         if (state!.Status != "Stopped")
-            await client.DisableAsync(outputStream);
-        await client.UpdateAsync(outputStream, query, true);
-        await client.EnableAsync(outputStream);
+            await client.DisableAsync(outputStream, cancellationToken: token);
+        await client.UpdateAsync(outputStream, query, true, cancellationToken: token);
+        await client.EnableAsync(outputStream, cancellationToken: token);
     }
 
-    public static async Task EnsureLookupProjection(this EventStoreProjectionManagementClient client, IProjectionRegister register, string category, string eventProperty, string outputStreamCategory)
+    public static async Task EnsureLookupProjection(this EventStoreProjectionManagementClient client, IProjectionRegister register, string category, string eventProperty, string outputStreamCategory, CancellationToken token = default)
     {
         string query =
             $"fromStreams(['$ce-{category}']).when( {{ \n    $any : function(s,e) {{ \n        if(e.body && e.body.{eventProperty}) {{\n            linkTo('{outputStreamCategory}-' + e.body.{eventProperty}, e) \n        }}\n        \n    }}\n}});";
         if ((await register.Get(outputStreamCategory)) != null)
             await Update(client, outputStreamCategory, query);
         else
         {
-            await client.CreateContinuousAsync(outputStreamCategory, query, false);
-            await client.DisableAsync(outputStreamCategory);
-            await client.UpdateAsync(outputStreamCategory, query, true);
-            await client.EnableAsync(outputStreamCategory);
+            await client.CreateContinuousAsync(outputStreamCategory, query, false, cancellationToken: token);
+            await client.DisableAsync(outputStreamCategory, cancellationToken: token);
+            await client.UpdateAsync(outputStreamCategory, query, true, cancellationToken: token);
+            await client.EnableAsync(outputStreamCategory, cancellationToken: token);
         }
     }
     public static async Task EnsureJoinProjection(this EventStoreProjectionManagementClient client,
-        string outputStream, IProjectionRegister register, IEnumerable<string> eventTypes)
+        string outputStream, IProjectionRegister register, IEnumerable<string> eventTypes, CancellationToken token = default)
     {
         var query = CreateQuery(outputStream, eventTypes);
 
         if ((await register.Get(outputStream)) != null)
-            await Update(client, outputStream, query);
+            await Update(client, outputStream, query, token);
         else
         {
-            await client.CreateContinuousAsync(outputStream, query, false);
-            await client.DisableAsync(outputStream);
-            await client.UpdateAsync(outputStream, query, true);
-            await client.EnableAsync(outputStream);
+            await client.CreateContinuousAsync(outputStream, query, false, cancellationToken: token);
+            await client.DisableAsync(outputStream, cancellationToken: token);
+            await client.UpdateAsync(outputStream, query, true, cancellationToken: token);
+            await client.EnableAsync(outputStream, cancellationToken: token);
         }
     }