Merge branch 'develop'

Author: rasmus

RELEASE_NOTES.md

@@ -1,4 +1,11 @@
-### New in 0.58 (not released yet)
+### New in 0.59 (not released yet)
+
+* Fix: Commands are now correctly published when no events are emitted from a saga
+  after handling a domain event
+* Minor fix: Updated name of Primary Key for MSSQL Snapshot Store to be different
+  from MSSQL Event Store, so both can be used in the same database without conflicts
+
+### New in 0.58.3377 (released 2018-05-15)
 
 * Minor fix: Corrected log in `CommandBus` regarding events emitted due to command
 

Source/EventFlow.MsSql/SnapshotStores/Scripts/0001 - Create EventFlowSnapshots.sql

@@ -7,7 +7,7 @@ BEGIN
 		[AggregateSequenceNumber] [int] NOT NULL,
 		[Data] [nvarchar](MAX) NOT NULL,
 		[Metadata] [nvarchar](MAX) NOT NULL,
-		CONSTRAINT [PK_EventFlow] PRIMARY KEY CLUSTERED
+		CONSTRAINT [PK_EventFlowSnapshots] PRIMARY KEY CLUSTERED
 		(
 			[Id] ASC
 		)
@@ -19,4 +19,4 @@ BEGIN
 		[AggregateId] ASC,
 		[AggregateSequenceNumber] ASC
 	)
-END
+END

Source/EventFlow.TestHelpers/Aggregates/Sagas/ThingySaga.cs

@@ -110,4 +110,4 @@ public void Apply(ThingySagaCompletedEvent aggregateEvent)
             Complete();
         }
     }
-}
+}

Source/EventFlow.TestHelpers/Test.cs

@@ -79,10 +79,10 @@ protected T Inject<T>(T instance)
             return instance;
         }
 
-        protected Mock<T> InjectMock<T>()
+        protected Mock<T> InjectMock<T>(params object[] args)
             where T : class
         {
-            var mock = new Mock<T>();
+            var mock = new Mock<T>(args);
             Fixture.Inject(mock.Object);
             return mock;
         }

Source/EventFlow.Tests/UnitTests/Sagas/AggregateSagas/SagaAggregateStoreTests.cs

@@ -23,14 +23,15 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using EventFlow.Aggregates;
 using EventFlow.Core;
 using EventFlow.Core.Caching;
-using EventFlow.Sagas;
 using EventFlow.Sagas.AggregateSagas;
 using EventFlow.TestHelpers;
+using EventFlow.TestHelpers.Aggregates.Events;
 using EventFlow.TestHelpers.Aggregates.Sagas;
 using Moq;
 using NUnit.Framework;
@@ -40,26 +41,25 @@ namespace EventFlow.Tests.UnitTests.Sagas.AggregateSagas
     [Category(Categories.Unit)]
     public class SagaAggregateStoreTests : TestsFor<SagaAggregateStore>
     {
+        private ThingySagaId _thingySagaId;
+        private Mock<ThingySaga> _thingySaga;
+
         [SetUp]
         public void SetUp()
         {
+            _thingySagaId = A<ThingySagaId>();
+            _thingySaga = InjectMock<ThingySaga>(_thingySagaId);
+
             Inject<IMemoryCache>(A<MemoryCache>());
         }
 
         [Test]
         public async Task AggregateStore_UpdateAsync_IsInvoked()
         {
             // Arrange
-            var aggregateStoreMock = InjectMock<IAggregateStore>();
             var thingySagaId = A<ThingySagaId>();
             var sourceId = A<SourceId>();
-            aggregateStoreMock
-                .Setup(s => s.UpdateAsync(
-                    thingySagaId,
-                    sourceId,
-                    It.IsAny<Func<ThingySaga, CancellationToken, Task>>(),
-                    It.IsAny<CancellationToken>()))
-                .ReturnsAsync(new List<IDomainEvent>());
+            var aggregateStoreMock = ArrangeAggregateStoreMock(thingySagaId, sourceId, A<bool>());
 
             // Act
             await Sut.UpdateAsync(
@@ -78,5 +78,50 @@ await Sut.UpdateAsync(
                     It.IsAny<CancellationToken>()),
                 Times.Once);
         }
+
+        [TestCase(true)]
+        [TestCase(false)]
+        public async Task CommandBus_PublishAsync_IsInvoked(bool eventsAreEmitted)
+        {
+            // Arrange
+            var thingySagaId = A<ThingySagaId>();
+            var sourceId = A<SourceId>();
+            ArrangeAggregateStoreMock(thingySagaId, sourceId, eventsAreEmitted);
+
+            // Act
+            await Sut.UpdateAsync(
+                thingySagaId,
+                typeof(ThingySaga),
+                sourceId,
+                (s, c) => Task.FromResult(0),
+                CancellationToken.None);
+
+            // Assert
+            _thingySaga.Verify(
+                s => s.PublishAsync(It.IsAny<ICommandBus>(), It.IsAny<CancellationToken>()),
+                Times.AtLeastOnce);
+        }
+
+        private Mock<IAggregateStore> ArrangeAggregateStoreMock(
+            ThingySagaId thingySagaId,
+            ISourceId sourceId,
+            bool returnsDomainEvents)
+        {
+            var aggregateStoreMock = InjectMock<IAggregateStore>();
+            var domainEvents = ManyDomainEvents<ThingyPingEvent>(returnsDomainEvents ? 3 : 0)
+                .Cast<IDomainEvent>()
+                .ToList();
+
+            aggregateStoreMock
+                .Setup(s => s.UpdateAsync(
+                    thingySagaId,
+                    sourceId,
+                    It.IsAny<Func<ThingySaga, CancellationToken, Task>>(),
+                    It.IsAny<CancellationToken>()))
+                .Callback<ThingySagaId, ISourceId, Func<ThingySaga, CancellationToken, Task>, CancellationToken>(
+                    (id, sid, f, c) => f(_thingySaga.Object, CancellationToken.None).Wait(c))
+                .ReturnsAsync(domainEvents);
+            return aggregateStoreMock;
+        }
     }
 }

Source/EventFlow/Sagas/AggregateSagas/AggregateSaga.cs

@@ -64,13 +64,15 @@ protected void Publish<TCommandAggregate, TCommandAggregateIdentity, TExecutionR
             ? SagaState.Completed
             : IsNew ? SagaState.New : SagaState.Running;
 
-        public async Task PublishAsync(ICommandBus commandBus, CancellationToken cancellationToken)
+        public virtual async Task PublishAsync(ICommandBus commandBus, CancellationToken cancellationToken)
         {
-            foreach (var unpublishedCommand in _unpublishedCommands.ToList())
+            var commandsToPublish = _unpublishedCommands.ToList();
+            _unpublishedCommands.Clear();
+
+            foreach (var unpublishedCommand in commandsToPublish)
             {
                 await unpublishedCommand(commandBus, cancellationToken).ConfigureAwait(false);
-                _unpublishedCommands.Remove(unpublishedCommand);
             }
         }
     }
-}
+}

Source/EventFlow/Sagas/AggregateSagas/SagaAggregateStore.cs

@@ -65,7 +65,7 @@ public override async Task<ISaga> UpdateAsync(
                     cancellationToken)
                 .ConfigureAwait(false);
 
-            var domainEvents = await storeAggregateSagaAsync(
+            await storeAggregateSagaAsync(
                     this,
                     sagaId,
                     sourceId,
@@ -77,7 +77,7 @@ public override async Task<ISaga> UpdateAsync(
                     cancellationToken)
                 .ConfigureAwait(false);
 
-            if (!domainEvents.Any())
+            if (saga is null)
             {
                 return null;
             }

appveyor.yml

@@ -1,7 +1,7 @@
 init:
   - git config --global core.autocrlf input
 
-version: 0.58.{build}
+version: 0.59.{build}
 
 skip_tags: true