ReadThroughAggregateCache indexes by both type and ID (#126)

Author: joshkempner

src/ReactiveDomain.Foundation.Tests/when_using_caching_repository.cs

@@ -1,6 +1,4 @@
 using System;
-using System.Threading;
-using Newtonsoft.Json;
 using ReactiveDomain.Foundation.StreamStore;
 using ReactiveDomain.Messaging;
 using ReactiveDomain.Testing.EventStore;
@@ -28,7 +26,6 @@ public when_using_caching_repository()
         [Fact]
         public void can_handle_repeated_and_cross_updates()
         {
-
             var id = Guid.NewGuid();
             ICorrelatedMessage source = MessageBuilder.New(() => new CreateAccount(id));
             var cachedAccount = new Account(id, source);
@@ -82,6 +79,23 @@ public void can_handle_repeated_and_cross_updates()
             Assert.Equal(8, cachedAccount3.Balance); //cachedAccount is a pointer to same object as acct2/3/4 due to cache
             Assert.Equal(8, cachedAccount4.Balance); //cachedAccount is a pointer to same object as acct2/3/4 due to cache
         }
+
+        [Fact]
+        public void can_cache_aggregates_with_different_types_and_same_id()
+        {
+            var accountId = Guid.NewGuid();
+            ICorrelatedMessage source = MessageBuilder.New(() => new CreateAccount(accountId));
+            var account = new Account(accountId, source);
+            _cachingRepo.Save(account);
+
+            ICorrelatedMessage source2 = MessageBuilder.New(() => new CreateAccount(accountId));
+            var other = new OtherAggregateType(accountId, source2);
+            _cachingRepo.Save(other);
+
+            var cachedAccount = _cachingRepo.GetById<Account>(accountId);
+            var cachedOther = _cachingRepo.GetById<OtherAggregateType>(accountId);
+        }
+
         [Fact]
         public void can_remove_by_id_from_the_cache()
         {
@@ -98,7 +112,7 @@ public void can_remove_by_id_from_the_cache()
             var cacheCopy = _cachingRepo.GetById<Account>(id);
             Assert.Equal(12, cacheCopy.Balance);
 
-            _cachingRepo.ClearCache(id);
+            _cachingRepo.ClearCache<Account>(id);
 
             var persistedCopy = _cachingRepo.GetById<Account>(id);
             Assert.Equal(5, persistedCopy.Balance);
@@ -130,14 +144,15 @@ private Account()
             }
             public Account(Guid id, ICorrelatedMessage source) : this()
             {
-                Ensure.NotEmptyGuid(id, "id");
+                Ensure.NotEmptyGuid(id, nameof(id));
                 Raise(new AccountCreated(id));
             }
             public void Credit(uint amount)
             {
                 Raise(new AccountCredited(Id, amount));
             }
         }
+
         public class CreateAccount : Command
         {
             public readonly Guid AccountId;
@@ -146,6 +161,7 @@ public CreateAccount(Guid accountId)
                 AccountId = accountId;
             }
         }
+
         public class AccountCreated : IMessage
         {
             public Guid MsgId { get; private set; }
@@ -156,6 +172,7 @@ public AccountCreated(Guid accountId)
                 AccountId = accountId;
             }
         }
+
         public class CreditAccount : Command
         {
             public readonly Guid AccountId;
@@ -166,6 +183,7 @@ public CreditAccount(Guid accountId, uint amount)
                 Amount = amount;
             }
         }
+
         public class AccountCredited : IMessage
         {
             public Guid MsgId { get; private set; }
@@ -179,5 +197,40 @@ public AccountCredited(Guid accountId, uint amount)
                 Amount = amount;
             }
         }
+
+        public class OtherAggregateType : EventDrivenStateMachine
+        {
+            private OtherAggregateType()
+            {
+                Register<OtherAggregateCreated>(e => Id = e.AggregateId);
+            }
+            public OtherAggregateType(Guid id, ICorrelatedMessage source) : this()
+            {
+                Ensure.NotEmptyGuid(id, nameof(id));
+                Raise(new OtherAggregateCreated(id));
+            }
+        }
+
+        public class CreateOtherAggregate : Command
+        {
+            public readonly Guid AggregateId;
+
+            public CreateOtherAggregate(Guid aggregateId)
+            {
+                AggregateId = aggregateId;
+            }
+        }
+
+        public class OtherAggregateCreated : IMessage
+        {
+            public Guid MsgId { get; }
+            public readonly Guid AggregateId;
+
+            public OtherAggregateCreated(Guid aggregateId)
+            {
+                MsgId = Guid.NewGuid();
+                AggregateId = aggregateId;
+            }
+        }
     }
 }

src/ReactiveDomain.Foundation/StreamStore/CachingRepository.cs

@@ -37,8 +37,8 @@ public void Delete(IEventSource aggregate) {
         public void HardDelete(IEventSource aggregate) {
             _cache.HardDelete(aggregate);
         }
-        public bool ClearCache(Guid id) {
-            return _cache.Remove(id);
+        public bool ClearCache<TAggregate>(Guid id) {
+            return _cache.Remove<TAggregate>(id);
         }
         public void ClearCache() {
             _cache.Clear();

src/ReactiveDomain.Foundation/StreamStore/IAggregateCache.cs

@@ -1,13 +1,14 @@
 using System;
 
-namespace ReactiveDomain.Foundation.StreamStore {
+namespace ReactiveDomain.Foundation.StreamStore
+{
     /// <summary>
     /// While it might seem more natural to save and restore the event set behind the aggregate,
     /// this cache stores only the collapsed state in the aggregate  
     /// </summary>
-    public interface IAggregateCache:IRepository, IDisposable
+    public interface IAggregateCache : IRepository, IDisposable
     {
-        bool Remove(Guid id);
+        bool Remove<TAggregate>(Guid id);
         void Clear();
     }
 }

src/ReactiveDomain.Foundation/StreamStore/ReadThroughAggregateCache.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections;
 using System.Collections.Generic;
 
 namespace ReactiveDomain.Foundation.StreamStore
@@ -20,7 +19,7 @@ public class ReadThroughAggregateCache : IAggregateCache, IDisposable
     {
 
         private readonly IRepository _baseRepository;
-        private readonly Dictionary<Guid, IEventSource> _knownAggregates = new Dictionary<Guid, IEventSource>();
+        private readonly Dictionary<(Type type, Guid id), IEventSource> _knownAggregates = new Dictionary<(Type type, Guid id), IEventSource>();
         public ReadThroughAggregateCache(IRepository baseRepository) {
             _baseRepository = baseRepository;
         }
@@ -31,32 +30,32 @@ public bool TryGetById<TAggregate>(Guid id, out TAggregate aggregate, int versio
             }
             catch (Exception) {
                 aggregate = null;
-                Remove(id);
+                Remove<TAggregate>(id);
                 return false;
             }
         }
 
         public TAggregate GetById<TAggregate>(Guid id, int version = int.MaxValue) where TAggregate : class, IEventSource {
-            if (_knownAggregates.TryGetValue(id, out var cached)) {
+            if (_knownAggregates.TryGetValue((typeof(TAggregate), id), out var cached)) {
                 var agg = (TAggregate) cached;
                 Update(ref agg, version);
                 return (TAggregate)cached;
             }
 
             var aggregate = _baseRepository.GetById<TAggregate>(id, version);
-            _knownAggregates.Add(id, aggregate);
+            _knownAggregates.Add((typeof(TAggregate), id), aggregate);
             return aggregate;
         }
 
         public void Update<TAggregate>(ref TAggregate aggregate, int version = int.MaxValue) where TAggregate : class, IEventSource {
             if (aggregate == null) throw new ArgumentNullException(nameof(aggregate));
             if (aggregate.ExpectedVersion == version) return;
 
-            _knownAggregates.TryGetValue(aggregate.Id, out var cached);
+            _knownAggregates.TryGetValue((typeof(TAggregate), aggregate.Id), out var cached);
 
             if (cached == null) {
                 _baseRepository.Update(ref aggregate, version);
-                UpdateCache(aggregate);
+                UpdateCache<TAggregate>(aggregate);
                 return;
             }
 
@@ -74,34 +73,35 @@ public void Update<TAggregate>(ref TAggregate aggregate, int version = int.MaxVa
             //cache is ahead of item, but behind requested version
             aggregate = (TAggregate)cached;
             _baseRepository.Update(ref aggregate, version);
-            UpdateCache(aggregate);
+            UpdateCache<TAggregate>(aggregate);
         }
 
 
 
-        private void UpdateCache(IEventSource aggregate) {
-            if (_knownAggregates.TryGetValue(aggregate.Id, out var cached)) {
+        private void UpdateCache<TAggregate>(IEventSource aggregate) {
+            if (_knownAggregates.TryGetValue((typeof(TAggregate), aggregate.Id), out var cached)) {
                 if (cached.ExpectedVersion < aggregate.ExpectedVersion) {
-                    _knownAggregates[aggregate.Id] = aggregate;
+                    _knownAggregates[(typeof(TAggregate), aggregate.Id)] = aggregate;
                 }
             }
             else {
-                _knownAggregates.Add(aggregate.Id, aggregate);
+                _knownAggregates.Add((typeof(TAggregate), aggregate.Id), aggregate);
             }
         }
 
         public void Save(IEventSource aggregate) {
+            var type = aggregate.GetType();
             try {
                 _baseRepository.Save(aggregate);
-                if (!_knownAggregates.ContainsKey(aggregate.Id)) {
-                    _knownAggregates.Add(aggregate.Id, aggregate);
+                if (!_knownAggregates.ContainsKey((type, aggregate.Id))) {
+                    _knownAggregates.Add((type, aggregate.Id), aggregate);
                 }
                 else {
-                    _knownAggregates[aggregate.Id] = aggregate;
+                    _knownAggregates[(type, aggregate.Id)] = aggregate;
                 }
             }
             catch {
-                _knownAggregates.Remove(aggregate.Id);
+                _knownAggregates.Remove((type, aggregate.Id));
             }
 
 
@@ -113,7 +113,7 @@ public void Save(IEventSource aggregate) {
         /// <param name="aggregate">The aggregate to be deleted.</param>
         public void Delete(IEventSource aggregate) {
             _baseRepository.Delete(aggregate);
-            _knownAggregates.Remove(aggregate.Id);
+            _knownAggregates.Remove((aggregate.GetType(), aggregate.Id));
         }
 
         /// <summary>
@@ -122,11 +122,11 @@ public void Delete(IEventSource aggregate) {
         /// <param name="aggregate">The aggregate to be deleted.</param>
         public void HardDelete(IEventSource aggregate) {
             _baseRepository.HardDelete(aggregate);
-            _knownAggregates.Remove(aggregate.Id);
+            _knownAggregates.Remove((aggregate.GetType(), aggregate.Id));
         }
 
-        public bool Remove(Guid id) {
-            return _knownAggregates.Remove(id);
+        public bool Remove<TAggregate>(Guid id) {
+            return _knownAggregates.Remove((typeof(TAggregate), id));
         }
         public void Clear() {
             _knownAggregates.Clear();