Green tests,

saga and aggregate actors can notify about events persisted

Author: andrey.leskov

GridDomain.Domain.Tests/Sagas/InstanceSagas/Given_saga_When_publishing_any_of_start_messages.cs

@@ -9,35 +9,34 @@
 
 namespace GridDomain.Tests.Sagas.InstanceSagas
 {
-
-
     [TestFixture]
-    class Given_saga_When_publishing_any_of_start_messages : Given_saga_When_publishing_start_messages
+    class Given_saga_When_publishing_any_of_start_messages : SoftwareProgrammingInstanceSagaTest
     {
-        private static Guid SagaId = Guid.NewGuid();
+        private static readonly Guid SagaId = Guid.NewGuid();
+        private SagaDataAggregate<SoftwareProgrammingSagaData> _sagaData;
 
-        public Given_saga_When_publishing_any_of_start_messages():
-            base(SagaId, new SleptWellEvent(Guid.NewGuid(), Guid.NewGuid(), SagaId))
+        [OneTimeSetUp]
+        public void When_publishing_start_message()
         {
-            
+            GridNode.NewDebugWaiter(Timeout)
+                    .Expect<SagaCreatedEvent<SoftwareProgrammingSagaData>>()
+                    .Create()
+                    .Publish(new SleptWellEvent(Guid.NewGuid(), Guid.NewGuid(), SagaId))
+                    .Wait();
+
+            _sagaData = LoadAggregate<SagaDataAggregate<SoftwareProgrammingSagaData>>(SagaId);
         }
 
         [Then]
         public void Saga_data_is_not_null()
         {
-            Assert.NotNull(SagaData.Data);
+            Assert.NotNull(_sagaData.Data);
         }
 
-
         [Then]
         public void Saga_has_correct_id()
         {
-            Assert.AreEqual(_sagaId, SagaData.Id);
-        }
-
-        protected override IExpectBuilder<AnyMessagePublisher> ConfigureWait(IMessageWaiter<AnyMessagePublisher> waiter)
-        {
-            return waiter.Expect<SagaCreatedEvent<SoftwareProgrammingSagaData>>();
+            Assert.AreEqual(SagaId, _sagaData.Id);
         }
     }
 }

GridDomain.Domain.Tests/Sagas/InstanceSagas/Given_saga_When_publishing_several_start_messages.cs

@@ -1,40 +1,50 @@
 using System;
+using System.Threading;
+using Akka.Actor;
+using Akka.Persistence;
 using GridDomain.CQRS;
 using GridDomain.EventSourcing.Sagas;
 using GridDomain.EventSourcing.Sagas.InstanceSagas;
+using GridDomain.Node.Actors;
 using GridDomain.Tests.Framework;
 using GridDomain.Tests.Sagas.SoftwareProgrammingDomain.Events;
 using NUnit.Framework;
 
 namespace GridDomain.Tests.Sagas.InstanceSagas
 {
     [TestFixture]
-    class Given_saga_When_publishing_several_start_messages : Given_saga_When_publishing_start_messages
+    class Given_saga_When_publishing_several_start_messages : SoftwareProgrammingInstanceSagaTest
     {
         private static readonly Guid SagaId = Guid.NewGuid();
-        private static  GotTiredEvent firstMessage;
-        private static  SleptWellEvent secondMessage;
-        private static object[] GetMessages(Guid sagaId)
-        {
-            firstMessage = new GotTiredEvent(Guid.NewGuid(), Guid.NewGuid(), Guid.NewGuid(), sagaId);
-            secondMessage = new SleptWellEvent(Guid.NewGuid(), Guid.NewGuid(), sagaId);
-
-            return new object[]
-            {
-                firstMessage,
-                secondMessage
-            };
-        }
+        private static  SleptWellEvent secondStartMessage;
+        private SagaDataAggregate<SoftwareProgrammingSagaData> SagaData;
 
-        public Given_saga_When_publishing_several_start_messages(): base(SagaId,GetMessages(SagaId))
+        [OneTimeSetUp]
+        public void When_publishing_start_message()
         {
+           var waiter = GridNode.NewDebugWaiter(Timeout);
+
+           secondStartMessage = new SleptWellEvent(Guid.NewGuid(), Guid.NewGuid(), SagaId);
+
+           waiter.Expect<SagaCreatedEvent<SoftwareProgrammingSagaData>>()
+                 .Create()
+                 .Publish(new GotTiredEvent(Guid.NewGuid(), Guid.NewGuid(), Guid.NewGuid(), SagaId),
+                          secondStartMessage)
+                 .Wait();
+
+           LookupInstanceSagaActor<SoftwareProgrammingSaga, SoftwareProgrammingSagaData>(SagaId)
+                                                       .Tell(NotifyOnPersistenceEvents.Instance);
 
+           int count = 2;
+           FishForMessage<Persisted>(m => ++count >=2);
+
+           SagaData = LoadAggregate<SagaDataAggregate<SoftwareProgrammingSagaData>>(SagaId);
         }
 
         [Then]
         public void Saga_reinitialized_from_last_start_message()
         {
-            Assert.AreEqual(secondMessage.SofaId, SagaData.Data.SofaId);
+            Assert.AreEqual(secondStartMessage.SofaId, SagaData.Data.SofaId);
         }
 
         [Then]
@@ -43,10 +53,5 @@ public void Saga_has_correct_state()
             var saga = new SoftwareProgrammingSaga();
             Assert.AreEqual(saga.Coding.Name, SagaData.Data.CurrentStateName);
         }
-
-        protected override IExpectBuilder<AnyMessagePublisher> ConfigureWait(IMessageWaiter<AnyMessagePublisher> waiter)
-        {
-            return waiter.Expect<SagaCreatedEvent<SoftwareProgrammingSagaData>>();
-        }
     }
 }

GridDomain.Node/Actors/AggregateActor.cs

@@ -98,6 +98,8 @@ private void ProcessAggregateEvents(ICommand command)
                          .With<FutureEventCanceledEvent>(Handle);
 
                 Publisher.Publish(e);
+
+                NotifyWatchers(new Persisted(e));
             });
 
             State.ClearUncommittedEvents();

GridDomain.Node/Actors/AggregatePersistedNotification.cs

@@ -2,15 +2,13 @@
 
 namespace GridDomain.Node.Actors
 {
-    public class AggregatePersistedNotification
+    public class Persisted
     {
-        public AggregatePersistedNotification(Guid id, Guid commandId)
+        public object Event { get;}
+
+        public Persisted(object @event)
         {
-            Id = id;
-            CommandId = commandId;
+            Event = @event;
         }
-
-        public Guid CommandId { get; }
-        public Guid Id { get; }
     }
 }

GridDomain.Node/Actors/EventSourcedActor.cs

@@ -40,7 +40,6 @@ public EventSourcedActor(IConstructAggregates aggregateConstructor,
                 Monitor.IncrementMessagesReceived();
                 DeleteSnapshots(SnapshotsPolicy.GetSnapshotsToDelete());
                 Become(Terminating);
-             //   StopNow(new DeleteSnapshotsSuccess(SnapshotSelectionCriteria.None));
             });
 
             Command<CheckHealth>(s => Sender.Tell(new HealthStatus(s.Payload)));
@@ -88,7 +87,7 @@ protected bool TrySaveSnapshot(object[] stateChange)
             return shouldSave;
         }
 
-        private void NotifyWatchers(object msg)
+        protected void NotifyWatchers(object msg)
         {
             foreach (var watcher in _persistenceWaiters)
                 watcher.Tell(msg);

GridDomain.Node/Actors/SagaActor.cs

@@ -88,7 +88,6 @@ private void ProcessSaga(object message)
             try
             {
                 Saga.Transit(message);
-               // SnapshotsPolicy.RefreshActivity();
             }
             catch (Exception ex)
             {
@@ -117,7 +116,11 @@ private void ProcessSagaCommands()
         private object[] ProcessSagaStateChange()
         {
             var stateChangeEvents = State.GetUncommittedEvents().Cast<object>().ToArray();
-            PersistAll(stateChangeEvents, e => Publisher.Publish(e));
+            PersistAll(stateChangeEvents, 
+                e => {
+                         Publisher.Publish(e);
+                         NotifyWatchers(new Persisted(e));
+                });
             State.ClearUncommittedEvents();
             return stateChangeEvents;
         }