Fixes threading issues with ReadModelBase (#140)

- Adds a `ReaderLock` property to `ReadModelBase` that can be used when reading the state of the read model.
- Adds a `Version` property to `ReadModelBase` that increments with the total number of messages passed to the read model. This can be used in tests to ensure the read model state.
- Fixes an issue in `ReadModelBase` that could cause events to be read twice if `Start` is called with a checkpoint that is already at the end of the stream.

Author: joshkempner

src/ReactiveDomain.Foundation.Tests/when_using_read_model_base.cs

@@ -11,14 +11,6 @@ public class when_using_read_model_base :
                     IHandle<when_using_read_model_base.ReadModelTestEvent>,
                     IClassFixture<StreamStoreConnectionFixture> {
 
-        private static IListener GetListener() {
-            return new QueuedStreamListener(
-                        nameof(when_using_read_model_base),
-                        _conn,
-                        Namer,
-                        Serializer);
-        }
-
         private static IStreamStoreConnection _conn;
         private static readonly IEventSerializer Serializer =
             new JsonMessageSerializer();
@@ -30,8 +22,7 @@ private static IListener GetListener() {
 
 
         public when_using_read_model_base(StreamStoreConnectionFixture fixture)
-                    : base(nameof(when_using_read_model_base), GetListener) {
-            //_conn = new MockStreamStoreConnection("mockStore");
+                    : base(nameof(when_using_read_model_base), new ConfiguredConnection(fixture.Connection, Namer, Serializer)) {
             _conn = fixture.Connection;
             _conn.Connect();
 

src/ReactiveDomain.Foundation.Tests/when_using_snapshot_read_model.cs

@@ -9,14 +9,8 @@
 namespace ReactiveDomain.Foundation.Tests {
     // ReSharper disable once InconsistentNaming
     public class when_using_snapshot_read_model : IClassFixture<StreamStoreConnectionFixture> {
-        private IListener GetListener() {
-            return new QueuedStreamListener(
-                nameof(when_using_read_model_base),
-                _conn,
-                _namer,
-                _serializer);
-        }
 
+        private readonly IConfiguredConnection _configuredConnection;
         private readonly IStreamStoreConnection _conn;
         private readonly IEventSerializer _serializer =
             new JsonMessageSerializer();
@@ -39,6 +33,7 @@ private void AppendEvents(
         public when_using_snapshot_read_model(StreamStoreConnectionFixture fixture) {
             _conn = fixture.Connection;
             _conn.Connect();
+            _configuredConnection = new ConfiguredConnection(_conn, _namer, _serializer);
 
             _aggId = Guid.NewGuid();
             _stream = _namer.GenerateForAggregate(typeof(SnapReadModelTestAggregate), _aggId);
@@ -48,7 +43,7 @@ public when_using_snapshot_read_model(StreamStoreConnectionFixture fixture) {
         }
         [Fact]
         public void can_get_snapshot_from_read_model() {
-            var rm = new TestSnapShotReadModel(_aggId, GetListener, null);
+            var rm = new TestSnapShotReadModel(_aggId, _configuredConnection, null);
             AssertEx.IsOrBecomesTrue(() => rm.Count == 10);
             var snapshot = rm.GetState();
 
@@ -68,7 +63,7 @@ public void can_apply_snapshot_to_read_model() {
                                 new List<Tuple<string, long>> { new Tuple<string, long>(_stream, 9) },
                                 new TestSnapShotReadModel.MyState { Count = 10, Sum = 20 });
 
-            var rm = new TestSnapShotReadModel(_aggId, GetListener, snapshot);
+            var rm = new TestSnapShotReadModel(_aggId, _configuredConnection, snapshot);
             AssertEx.IsOrBecomesTrue(() => rm.Count == 10);
             AssertEx.IsOrBecomesTrue(() => rm.Sum == 20);
             AppendEvents(1, _conn, _stream, 5);
@@ -78,15 +73,15 @@ public void can_apply_snapshot_to_read_model() {
         [Fact]
         [SuppressMessage("ReSharper", "AccessToDisposedClosure")]
         public void can_snapshot_and_recover_read_model() {
-            var rm = new TestSnapShotReadModel(_aggId, GetListener, null);
+            var rm = new TestSnapShotReadModel(_aggId, _configuredConnection, null);
             AssertEx.IsOrBecomesTrue(() => rm.Count == 10);
             AssertEx.IsOrBecomesTrue(() => rm.Sum == 20);
             AppendEvents(1, _conn, _stream, 5);
             AssertEx.IsOrBecomesTrue(() => rm.Count == 11, 1000);
             AssertEx.IsOrBecomesTrue(() => rm.Sum == 25);
             var snap = rm.GetState();
             rm.Dispose();
-            var rm2 = new TestSnapShotReadModel(_aggId, GetListener, snap);
+            var rm2 = new TestSnapShotReadModel(_aggId, _configuredConnection, snap);
             AssertEx.IsOrBecomesTrue(() => rm2.Count == 11, 1000);
             AssertEx.IsOrBecomesTrue(() => rm2.Sum == 25);
             AppendEvents(1, _conn, _stream, 5);
@@ -95,15 +90,15 @@ public void can_snapshot_and_recover_read_model() {
         }
         [Fact]
         public void can_only_restore_once() {
-            var rm = new TestSnapShotReadModel(_aggId, GetListener, null);
+            var rm = new TestSnapShotReadModel(_aggId, _configuredConnection, null);
             // ReSharper disable once AccessToDisposedClosure
             AssertEx.IsOrBecomesTrue(() => rm.Count == 10);
             var state = rm.GetState();
             rm.Dispose();
             //create while forcing double restore
             Assert.Throws<InvalidOperationException>(() => new TestSnapShotReadModel(
-                                                                    _aggId, 
-                                                                    GetListener, 
+                                                                    _aggId,
+                                                                    _configuredConnection, 
                                                                     state,
                                                                     forceDoubleRestoreError:true));
         }
@@ -114,7 +109,7 @@ public void can_restore_state_without_checkpoints() {
                 null, //no streams provided
                 new TestSnapShotReadModel.MyState { Count = 10, Sum = 20 });
 
-            var rm = new TestSnapShotReadModel(_aggId, GetListener, snapshot);
+            var rm = new TestSnapShotReadModel(_aggId, _configuredConnection, snapshot);
             AssertEx.IsOrBecomesTrue(() => rm.Count == 10);
             AssertEx.IsOrBecomesTrue(() => rm.Sum == 20);
 
@@ -135,7 +130,7 @@ public void can_restore_state_without_checkpoints() {
             //can still get complete snapshot
             var snap2 = rm.GetState();
             //works as expected
-            var rm2 = new TestSnapShotReadModel(_aggId, GetListener, snap2);
+            var rm2 = new TestSnapShotReadModel(_aggId, _configuredConnection, snap2);
             AssertEx.IsOrBecomesTrue(() => rm2.Count == 12, 1000);
             AssertEx.IsOrBecomesTrue(() => rm2.Sum == 30);
             AppendEvents(1, _conn, _stream, 5);
@@ -148,10 +143,10 @@ public sealed class TestSnapShotReadModel :
             IHandle<SnapReadModelTestEvent> {
             public TestSnapShotReadModel(
                     Guid aggId,
-                    Func<IListener> getListener,
+                    IConfiguredConnection configuredConnection,
                     ReadModelState snapshot,
                     bool forceDoubleRestoreError = false) :
-                base(nameof(TestSnapShotReadModel), getListener) {
+                base(nameof(TestSnapShotReadModel), configuredConnection) {
                 // ReSharper disable once RedundantTypeArgumentsOfMethod
                 EventStream.Subscribe<SnapReadModelTestEvent>(this);
 

src/ReactiveDomain.Foundation/StreamStore/ReadModelBase.cs

@@ -24,38 +24,50 @@ public abstract class ReadModelBase :
         public bool Idle => _queue.Idle;
 
         /// <summary>
-        /// Creates a read model using the provided Function to get a listener.
-        /// This is deprecated and will be removed in a future version of ReactiveDomain.
+        /// ReaderLock locks the event handler and can be used when reading the model 
+        /// to ensure model state is unchanged during read.
+        /// The lock should *not* be used in Handle methods as they are inside the lock already by default.
         /// </summary>
-        /// <param name="name">The name of the read model. Also used as the name of the listener.</param>
-        /// <param name="getListener">A Func to get a new <see cref="IListener"/>.</param>
-        protected ReadModelBase(string name, Func<IListener> getListener)
-        {
-            Ensure.NotNull(getListener, nameof(getListener));
-            _getListener = getListener;
-            _listeners = new List<IListener>();
-            _bus = new InMemoryBus($"{nameof(ReadModelBase)}:{name} bus", false);
-            _queue = new QueuedHandler(_bus, $"{nameof(ReadModelBase)}:{name} queue");
-            _queue.Start();
-        }
+        protected readonly object ReaderLock = new object();
+
+        /// <summary>
+        /// The version is equal to the number of messages passed to the read model.
+        /// The version is incremented after all handlers have been processed.
+        /// The number of handlers (including none) will not impact the version.
+        /// This can be used to ensure read model state for tests. This is *not*
+        /// the same as the version of any particular stream being read.
+        /// </summary>
+        public int Version { get; private set; }
 
         /// <summary>
         /// Creates a read model using the provided stream store connection. Reads existing events using a
         /// reader, then transitions to a listener for live events.
         /// </summary>
-        /// <param name="name">The name of the read model. Also used as the name of the listener and reader.</param>
+        /// <param name="name">The name of the read model. Also used as the names of the listener and reader.</param>
         /// <param name="connection">A connection to a stream store.</param>
         protected ReadModelBase(string name, IConfiguredConnection connection)
         {
             Ensure.NotNull(connection, nameof(connection));
-            _getListener = () => connection.GetListener(name);
             _getReader = () => connection.GetReader(name, Handle);
+            _getListener = () => connection.GetListener(name);
             _listeners = new List<IListener>();
             _bus = new InMemoryBus($"{nameof(ReadModelBase)}:{name} bus", false);
-            _queue = new QueuedHandler(_bus, $"{nameof(ReadModelBase)}:{name} queue");
+            _queue = new QueuedHandler(new AdHocHandler<IMessage>(DequeueMessage), $"{nameof(ReadModelBase)}:{name} queue");
             _queue.Start();
         }
 
+        /// <summary>
+        /// Every message handled by the read model will pass through here.
+        /// </summary>
+        private void DequeueMessage(IMessage message)
+        {
+            lock (ReaderLock)
+            {
+                _bus.Handle(message);
+                Version++;
+            }
+        }
+
         private IListener AddNewListener()
         {
             var l = _getListener();
@@ -98,7 +110,7 @@ public void Start(string stream, long? checkpoint = null, bool blockUntilLive =
                 using (var reader = _getReader())
                 {
                     reader.Read(stream, () => Idle, checkpoint);
-                    checkpoint = reader.Position;
+                    checkpoint = reader.Position ?? checkpoint;
                 }
             }
             AddNewListener().Start(stream, checkpoint, blockUntilLive, cancelWaitToken);

src/ReactiveDomain.Foundation/StreamStore/SnapshotReadModel.cs

@@ -9,8 +9,8 @@ public abstract class SnapshotReadModel : ReadModelBase {
 
         protected SnapshotReadModel(
                 string name,
-                Func<IListener> getListener)
-            : base(name, getListener) {
+                IConfiguredConnection connection)
+            : base(name, connection) {
         }
 
         protected virtual void Restore(

src/ReactiveDomain.IdentityStorage/ReadModels/SubjectsRm.cs

@@ -12,7 +12,7 @@ public class SubjectsRm :
         IHandle<SubjectMsgs.SubjectCreated>
     {
         public SubjectsRm(IConfiguredConnection conn)
-            : base(nameof(SubjectsRm), () => conn.GetListener(nameof(SubjectsRm)))
+            : base(nameof(SubjectsRm), conn)
         {
             //set handlers
             EventStream.Subscribe<SubjectMsgs.SubjectCreated>(this);

src/ReactiveDomain.IdentityStorage/ReadModels/UsersRm.cs

@@ -19,7 +19,7 @@ public class UsersRm :
         private readonly SourceCache<UserDTO, Guid> _allUsers = new SourceCache<UserDTO, Guid>(x => x.UserId);
 
         private readonly List<Guid> _userIds = new List<Guid>();
-        public UsersRm(IConfiguredConnection conn) : base(nameof(UsersRm), () => conn.GetListener(nameof(UsersRm)))
+        public UsersRm(IConfiguredConnection conn) : base(nameof(UsersRm), conn)
         {
             long? position;
             EventStream.Subscribe<UserMsgs.UserEvent>(this);

src/ReactiveDomain.IdentityStorage/Services/ClientStore.cs

@@ -20,7 +20,7 @@ public class ClientStore :
         IHandle<ClientMsgs.ClientSecretRemoved>
     {
 
-        public ClientStore(IConfiguredConnection conn) : base(nameof(ClientStore), () => conn.GetListener(nameof(ClientStore)))
+        public ClientStore(IConfiguredConnection conn) : base(nameof(ClientStore), conn)
         {
             long checkpoint;
 

src/ReactiveDomain.PolicyStorage/ReadModels/ApplicationRm.cs

@@ -29,7 +29,7 @@ public class ApplicationRm :
 
 
         public ApplicationRm(IConfiguredConnection conn)
-           : base(nameof(ApplicationRm), () => conn.GetListener(nameof(ApplicationRm)))
+           : base(nameof(ApplicationRm), conn)
         {
             //set handlers
             EventStream.Subscribe<ApplicationMsgs.ApplicationCreated>(this);

src/ReactiveDomain.PolicyStorage/ReadModels/FilteredPoliciesRM.cs

@@ -1,6 +1,5 @@
 using DynamicData;
 using ReactiveDomain.Foundation;
-using ReactiveDomain.Messaging;
 using ReactiveDomain.Messaging.Bus;
 using ReactiveDomain.Policy.Messages;
 using System;
@@ -28,7 +27,7 @@ public class FilteredPoliciesRM :
         private readonly Dictionary<Guid, RoleDTO> _roles = new Dictionary<Guid, RoleDTO>();
 
         public FilteredPoliciesRM(IConfiguredConnection conn, List<string> policyFilter = null)
-           : base(nameof(FilteredPoliciesRM), () => conn.GetListener(nameof(FilteredPoliciesRM)))
+           : base(nameof(FilteredPoliciesRM), conn)
         {
             if (policyFilter != null)
             {

src/ReactiveDomain.PolicyStorage/ReadModels/PolicyUserRm.cs

@@ -22,7 +22,7 @@ public class PolicyUserRm :
         /// Create a read model for getting information about Policy Users.
         /// </summary>
         public PolicyUserRm(IConfiguredConnection conn)
-            : base(nameof(PolicyUserRm), () => conn.GetListener(nameof(PolicyUserRm)))
+            : base(nameof(PolicyUserRm), conn)
         {
             //set handlers
             EventStream.Subscribe<PolicyUserMsgs.PolicyUserAdded>(this);