From 50aedad886a65149ac1ba3faa8c9518c0fd1da5d Mon Sep 17 00:00:00 2001
From: Tomas Lycken <tlycken@users.noreply.github.com>
Date: Wed, 25 Oct 2017 20:39:04 +0200
Subject: [PATCH] Lock event store per stream (#29)

---
 .../Infrastructure/EventStoreFixture.cs                    | 4 ++--
 .../EntityFrameworkEventStore.cs                           | 6 +++---
 src/RdbmsEventStore/IWriteLock.cs                          | 4 ++--
 src/RdbmsEventStore/WriteLock.cs                           | 7 ++++---
 4 files changed, 11 insertions(+), 10 deletions(-)

diff --git a/src/RdbmsEventStore.EntityFramework.Tests/Infrastructure/EventStoreFixture.cs b/src/RdbmsEventStore.EntityFramework.Tests/Infrastructure/EventStoreFixture.cs
index d9ec1c6..6bc415d 100644
--- a/src/RdbmsEventStore.EntityFramework.Tests/Infrastructure/EventStoreFixture.cs
+++ b/src/RdbmsEventStore.EntityFramework.Tests/Infrastructure/EventStoreFixture.cs
@@ -19,13 +19,13 @@ public EventStoreFixture()
             EventRegistry = new AssemblyEventRegistry(typeof(TEvent), type => type.Name, type => !type.Name.StartsWith("<>"));
             EventSerializer = new DefaultEventSerializer<TStreamId, TEvent, TPersistedEvent>(EventRegistry);
             EventFactory = new DefaultEventFactory<TStreamId, TEvent>();
-            WriteLock = new WriteLock();
+            WriteLock = new WriteLock<TStreamId>();
         }
 
         public IEventRegistry EventRegistry { get; protected set; }
         public IEventSerializer<TEvent, TPersistedEvent> EventSerializer { get; protected set; }
         public IEventFactory<TStreamId, TEvent> EventFactory { get; protected set; }
-        public IWriteLock WriteLock { get; protected set; }
+        public IWriteLock<TStreamId> WriteLock { get; protected set; }
 
         public EntityFrameworkEventStore<TId, TStreamId, TEventStoreContext, TEvent, TEventMetadata, TPersistedEvent> BuildEventStore<TEventStoreContext>(TEventStoreContext dbContext)
             where TEventStoreContext : DbContext, IEventDbContext<TPersistedEvent>
diff --git a/src/RdbmsEventStore.EntityFramework/EntityFrameworkEventStore.cs b/src/RdbmsEventStore.EntityFramework/EntityFrameworkEventStore.cs
index ea3bdda..5d8a6d7 100644
--- a/src/RdbmsEventStore.EntityFramework/EntityFrameworkEventStore.cs
+++ b/src/RdbmsEventStore.EntityFramework/EntityFrameworkEventStore.cs
@@ -17,10 +17,10 @@ public class EntityFrameworkEventStore<TId, TStreamId, TContext, TEvent, TEventM
     {
         private readonly TContext context;
         private readonly IEventFactory<TStreamId, TEvent> _eventFactory;
-        private readonly IWriteLock _writeLock;
+        private readonly IWriteLock<TStreamId> _writeLock;
         private readonly IEventSerializer<TEvent, TPersistedEvent> _serializer;
 
-        public EntityFrameworkEventStore(TContext context, IEventFactory<TStreamId, TEvent> eventFactory, IWriteLock writeLock, IEventSerializer<TEvent, TPersistedEvent> serializer)
+        public EntityFrameworkEventStore(TContext context, IEventFactory<TStreamId, TEvent> eventFactory, IWriteLock<TStreamId> writeLock, IEventSerializer<TEvent, TPersistedEvent> serializer)
         {
             this.context = context;
             _eventFactory = eventFactory;
@@ -49,7 +49,7 @@ public Task Append(TStreamId streamId, long versionBefore, object payload)
 
         public async Task Append(TStreamId streamId, long versionBefore, IEnumerable<object> payloads)
         {
-            using (await _writeLock.Aquire())
+            using (await _writeLock.Aquire(streamId))
             {
                 var highestVersionNumber = await context.Events
                     .Where(e => e.StreamId.Equals(streamId))
diff --git a/src/RdbmsEventStore/IWriteLock.cs b/src/RdbmsEventStore/IWriteLock.cs
index 3eb228e..73b463b 100644
--- a/src/RdbmsEventStore/IWriteLock.cs
+++ b/src/RdbmsEventStore/IWriteLock.cs
@@ -3,8 +3,8 @@
 
 namespace RdbmsEventStore
 {
-    public interface IWriteLock
+    public interface IWriteLock<in TStreamId>
     {
-        AwaitableDisposable<IDisposable> Aquire();
+        AwaitableDisposable<IDisposable> Aquire(TStreamId streamId);
     }
 }
\ No newline at end of file
diff --git a/src/RdbmsEventStore/WriteLock.cs b/src/RdbmsEventStore/WriteLock.cs
index 3a88120..3b9b2b7 100644
--- a/src/RdbmsEventStore/WriteLock.cs
+++ b/src/RdbmsEventStore/WriteLock.cs
@@ -1,12 +1,13 @@
 using System;
+using System.Collections.Concurrent;
 using Nito.AsyncEx;
 
 namespace RdbmsEventStore
 {
-    public class WriteLock : IWriteLock
+    public class WriteLock<TStreamId> : IWriteLock<TStreamId>
     {
-        private readonly AsyncLock _mutex = new AsyncLock();
+        private readonly ConcurrentDictionary<TStreamId, AsyncLock> _mutexes = new ConcurrentDictionary<TStreamId, AsyncLock>();
 
-        public AwaitableDisposable<IDisposable> Aquire() => _mutex.LockAsync();
+        public AwaitableDisposable<IDisposable> Aquire(TStreamId streamId) => _mutexes.GetOrAdd(streamId, id => new AsyncLock()).LockAsync();
     }
 }
\ No newline at end of file