Merge pull request #7 from corvansteijn/FixIdempotency

Fix idempotency

Author: dennisdoomen

Src/LiquidProjections.NHibernate/NHibernateProjector.cs

@@ -119,8 +119,13 @@ public async Task Handle(IReadOnlyList<Transaction> transactions, SubscriptionIn
             {
                 throw new ArgumentNullException(nameof(transactions));
             }
+            
+            long? lastCheckpoint = GetLastCheckpoint();
+            IEnumerable<IList<Transaction>> transactionBatches = transactions
+                .Where(t => (!lastCheckpoint.HasValue) || (t.Checkpoint > lastCheckpoint))
+                .InBatchesOf(BatchSize);
 
-            foreach (IList<Transaction> batch in transactions.InBatchesOf(BatchSize))
+            foreach (IList<Transaction> batch in transactionBatches)
             {
                 await ExecuteWithRetry(() => ProjectTransactionBatch(batch)).ConfigureAwait(false);
             }
@@ -150,16 +155,15 @@ private async Task ProjectTransactionBatch(IList<Transaction> batch)
             try
             {
                 using (ISession session = sessionFactory())
+                using (var tx = session.BeginTransaction())
                 {
-                    session.BeginTransaction();
-
                     foreach (Transaction transaction in batch)
                     {
                         await ProjectTransaction(transaction, session).ConfigureAwait(false);
                     }
 
                     StoreLastCheckpoint(session, batch.Last());
-                    session.Transaction.Commit();
+                    tx.Commit();
                 }
             }
             catch (ProjectionException projectionException)

Tests/LiquidProjections.NHibernate.Specs/NHibernateProjectorSpecs.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Collections.ObjectModel;
 using System.Linq;
 using System.Threading.Tasks;
 using Chill;
@@ -20,7 +21,7 @@ public class Given_a_sqlite_projector_with_an_in_memory_event_source :
         {
             protected EventMapBuilder<ProductCatalogEntry, string, NHibernateProjectionContext> Events;
             protected LruProjectionCache<ProductCatalogEntry, string> Cache;
-            protected Exception ProjectionException = null;
+            protected Exception ProjectionException;
             private readonly List<INHibernateChildProjector> children = new List<INHibernateChildProjector>();
 
             public Given_a_sqlite_projector_with_an_in_memory_event_source()
@@ -68,11 +69,13 @@ protected void StartProjecting()
                         {
                             ProjectionException = e;
                         }
-                    } 
+                    }
                 }, "");
             }
         }
 
+        #region Mapping Rules
+
         public class When_a_create_was_requested_but_the_database_already_contained_that_projection :
             Given_a_sqlite_projector_with_an_in_memory_event_source
         {
@@ -151,7 +154,6 @@ public void Then_it_should_throw()
             {
                 ProjectionException.Should()
                     .BeOfType<ProjectionException>();
-
             }
         }
 
@@ -422,7 +424,7 @@ public When_a_create_or_update_of_an_existing_cached_projection_is_requested()
 
                         session.Flush();
                     }
-                    
+
                     Cache.Add(new ProductCatalogEntry
                     {
                         Id = "c350E",
@@ -1108,6 +1110,10 @@ public void Then_it_should_not_do_anything()
             }
         }
 
+        #endregion
+
+        #region General 
+
         public class When_a_custom_state_key_is_set : Given_a_sqlite_projector_with_an_in_memory_event_source
         {
             public When_a_custom_state_key_is_set()
@@ -1167,9 +1173,9 @@ public When_the_projector_state_is_enriched()
                         StreamId = "Product1",
                         Events = new[]
                         {
-                             new EventEnvelope
-                             {
-                                 Body = new ProductAddedToCatalogEvent
+                            new EventEnvelope
+                            {
+                                Body = new ProductAddedToCatalogEvent
                                 {
                                     ProductKey = "c350E",
                                     Category = "Hybrid"
@@ -1283,10 +1289,13 @@ public void Then_it_should_use_the_header()
             }
         }
 
+        #endregion
+
+        #region Child Projectors
+
         public class When_there_is_a_child_projector :
             Given_a_sqlite_projector_with_an_in_memory_event_source
         {
-            private Transaction transaction2;
             private readonly List<ChildProjectionState> childProjectionStates = new List<ChildProjectionState>();
 
             public When_there_is_a_child_projector()
@@ -1338,7 +1347,7 @@ public When_there_is_a_child_projector()
                         }
                     };
 
-                    transaction2 = new Transaction
+                    var transaction2 = new Transaction
                     {
                         Events = new[]
                         {
@@ -1409,6 +1418,7 @@ private class ChildProjectionState
                 public bool Entry2Exists { get; set; }
             }
         }
+
         public class When_a_child_projector_has_its_own_cache :
             Given_a_sqlite_projector_with_an_in_memory_event_source
         {
@@ -1455,6 +1465,58 @@ public async Task It_should_add_the_item_to_the_cache()
             }
         }
 
+        #endregion
+
+        #region Exception Handling
+
+        public class When_any_exception_is_thrown :
+            Given_a_sqlite_projector_with_an_in_memory_event_source
+        {
+            public When_any_exception_is_thrown()
+            {
+                Given(() =>
+                {
+                    Events.Map<ProductAddedToCatalogEvent>()
+                        .AsCreateOf(@event => @event.ProductKey)
+                        .Using((p, @event) =>
+                        {
+                            p.Category = @event.Category;
+                            p.Name = @event.Name;
+                        });
+
+                    Events.Map<CategoryDiscontinuedEvent>().As((@event, context) => throw new InvalidOperationException());
+
+                    StartProjecting();
+                });
+
+                When(async () =>
+                {
+                    await The<MemoryEventSource>()
+                        .Write(new TransactionBuilder()
+                            .WithEvent(new ProductAddedToCatalogEvent
+                            {
+                                Category = "some category",
+                                Name = "some product",
+                                ProductKey = "some key",
+                            })
+                            .WithEvent(new CategoryDiscontinuedEvent
+                            {
+                                Category = "some category"
+                            })
+                            .Build());
+                });
+            }
+
+            [Fact]
+            public void Then_it_should_rollback_the_transaction()
+            {
+                using (var session = The<ISessionFactory>().OpenSession())
+                {
+                    session.QueryOver<ProductCatalogEntry>().List().Should().BeEmpty();
+                }
+            }
+        }
+
         public class When_event_handling_fails :
             Given_a_sqlite_projector_with_an_in_memory_event_source
         {
@@ -1581,6 +1643,60 @@ public void Then_it_should_succeed()
                 succeeded.Should().BeTrue();
             }
         }
+
+        public class When_retrying_a_transaction_that_was_already_handled :
+            Given_a_sqlite_projector_with_an_in_memory_event_source
+        {
+            private ReadOnlyCollection<Transaction> transactions;
+            private SubscriptionInfo info;
+
+            public When_retrying_a_transaction_that_was_already_handled()
+            {
+                Given(() =>
+                {
+                    Events.Map<ProductAddedToCatalogEvent>()
+                        .AsCreateOf(@event => @event.ProductKey)
+                        .Using((p, @event) =>
+                        {
+                            p.Category = @event.Category;
+                            p.Name = @event.Name;
+                        });
+                    
+                    
+                    Transaction transaction = new TransactionBuilder()
+                        .WithCheckpointNumber(5)
+                        .WithEvent(new ProductAddedToCatalogEvent
+                        {
+                            Category = "some category",
+                            Name = $"some product",
+                            ProductKey = $"some key",
+                        })
+                        .Build();
+                    
+                    transactions = new List<Transaction>
+                        {
+                            transaction
+                        }
+                        .AsReadOnly();
+                    info = new SubscriptionInfo();
+                    
+                    StartProjecting();
+                });
+
+                When(async () =>
+                {
+                    await Subject.Handle(transactions, info);
+                    await Subject.Handle(transactions, info);
+                }, DeferredExecution = true);
+            }
+
+            [Fact]
+            public void Then_it_should_not_handle_the_transactions()
+            {
+                WhenAction.ShouldNotThrow();
+            }
+        }
+
         public class When_a_projection_exception_occurs :
             Given_a_sqlite_projector_with_an_in_memory_event_source
         {
@@ -1594,12 +1710,12 @@ public When_a_projection_exception_occurs()
                     {
                         throw The<InvalidOperationException>();
                     });
-                    
+
                     Cache.Add(new ProductCatalogEntry
                     {
                         Id = "c350E"
                     });
-                  
+
                     StartProjecting();
                 });
 
@@ -1625,6 +1741,7 @@ public void Then_it_should_completely_clear_the_cache()
                 Cache.CurrentCount.Should().Be(0);
             }
         }
+
         public class When_a_database_exception_occurs : Given_a_sqlite_projector_with_an_in_memory_event_source
         {
             public When_a_database_exception_occurs()
@@ -1634,7 +1751,7 @@ public When_a_database_exception_occurs()
                     Events.Map<ProductAddedToCatalogEvent>()
                         .AsCreateOf(e => e.ProductKey)
                         .Using((p, e) => p.Name = e.Name);
-                    
+
                     using (ISession session = The<ISessionFactory>().OpenSession())
                     {
                         var entry = new ProductCatalogEntry
@@ -1647,12 +1764,12 @@ public When_a_database_exception_occurs()
                         session.Save(entry);
                         session.Flush();
                     }
-                    
+
                     Cache.Add(new ProductCatalogEntry
                     {
                         Id = "SomeOtherCachedId"
                     });
-                  
+
                     StartProjecting();
                 });
 
@@ -1666,7 +1783,6 @@ await The<MemoryEventSource>().Write(new ProductAddedToCatalogEvent
                             Category = "Hybrids",
                             Name = "DuplicateNameThatWillThrowUniqueConstraintException"
                         });
-
                     }
                     catch
                     {
@@ -1681,6 +1797,8 @@ public void Then_it_should_completely_clear_the_cache()
                 Cache.CurrentCount.Should().Be(0);
             }
         }
+
+        #endregion
     }
 
     #region Supporting Types
@@ -1743,4 +1861,43 @@ public class CategoryDiscontinuedEvent
     }
 
     #endregion
+}
+
+namespace LiquidProjections.NHibernate.Specs.NHibernateProjectorSpecs
+{
+    public class TransactionBuilder
+    {
+        private readonly List<object> events = new List<object>();
+        private long? checkpointNumber;
+
+        public TransactionBuilder WithEvent(object @event)
+        {
+            events.Add(@event);
+            return this;
+        }
+
+        public Transaction Build()
+        {
+            Transaction transaction = new Transaction
+            {
+                Events = events.Select(e => new EventEnvelope
+                {
+                    Body = e
+                }).ToArray()
+            };
+
+            if (checkpointNumber.HasValue)
+            {
+                transaction.Checkpoint = checkpointNumber.Value;
+            }
+
+            return transaction;
+        }
+
+        public TransactionBuilder WithCheckpointNumber(long checkpointNumber)
+        {
+            this.checkpointNumber = checkpointNumber;
+            return this;
+        }
+    }
 }