Merge branch 'test2'

src/Squidex.Infrastructure.GetEventStore/CQRS/Events/GetEventStore.cs

@@ -80,10 +80,22 @@ namespace Squidex.Infrastructure.CQRS.Events
             return result;
         }
 
-        public async Task AppendEventsAsync(Guid commitId, string streamName, int expectedVersion, ICollection<EventData> events)
+        public Task AppendEventsAsync(Guid commitId, string streamName, ICollection<EventData> events)
+        {
+            return AppendEventsInternalAsync(streamName, ExpectedVersion.Any, events);
+        }
+
+        public Task AppendEventsAsync(Guid commitId, string streamName, int expectedVersion, ICollection<EventData> events)
+        {
+            Guard.GreaterEquals(expectedVersion, -1, nameof(expectedVersion));
+
+            return AppendEventsInternalAsync(streamName, expectedVersion, events);
+        }
+
+        private async Task AppendEventsInternalAsync(string streamName, long expectedVersion, ICollection<EventData> events)
         {
-            Guard.NotNull(events, nameof(events));
             Guard.NotNullOrEmpty(streamName, nameof(streamName));
+            Guard.NotNull(events, nameof(events));
 
             if (events.Count == 0)
             {

src/Squidex.Infrastructure.MongoDb/CQRS/Events/MongoEventStore.cs

@@ -25,6 +25,8 @@ namespace Squidex.Infrastructure.CQRS.Events
 {
     public class MongoEventStore : MongoRepositoryBase<MongoEventCommit>, IEventStore
     {
+        private const long AnyVersion = long.MinValue;
+        private const int MaxAttempts = 20;
         private static readonly BsonTimestamp EmptyTimestamp = new BsonTimestamp(0);
         private static readonly FieldDefinition<MongoEventCommit, BsonTimestamp> TimestampField = Fields.Build(x => x.Timestamp);
         private static readonly FieldDefinition<MongoEventCommit, long> EventsCountField = Fields.Build(x => x.EventsCount);
@@ -50,9 +52,10 @@ namespace Squidex.Infrastructure.CQRS.Events
             return new MongoCollectionSettings { ReadPreference = ReadPreference.Primary, WriteConcern = WriteConcern.WMajority };
         }
 
-        protected override Task SetupCollectionAsync(IMongoCollection<MongoEventCommit> collection)
+        protected override async Task SetupCollectionAsync(IMongoCollection<MongoEventCommit> collection)
         {
-            return collection.Indexes.CreateOneAsync(Index.Ascending(x => x.EventStreamOffset).Ascending(x => x.EventStream), new CreateIndexOptions { Unique = true });
+            await collection.Indexes.CreateOneAsync(Index.Ascending(x => x.Timestamp).Ascending(x => x.EventStream));
+            await collection.Indexes.CreateOneAsync(Index.Ascending(x => x.EventStream).Descending(x => x.EventStreamOffset), new CreateIndexOptions { Unique = true });
         }
 
         public IEventSubscription CreateSubscription(string streamFilter = null, string position = null)
@@ -107,49 +110,65 @@ namespace Squidex.Infrastructure.CQRS.Events
             }, cancellationToken);
         }
 
-        public async Task AppendEventsAsync(Guid commitId, string streamName, int expectedVersion, ICollection<EventData> events)
+        public Task AppendEventsAsync(Guid commitId, string streamName, ICollection<EventData> events)
+        {
+            return AppendEventsInternalAsync(commitId, streamName, AnyVersion, events);
+        }
+
+        public Task AppendEventsAsync(Guid commitId, string streamName, int expectedVersion, ICollection<EventData> events)
+        {
+            Guard.GreaterEquals(expectedVersion, -1, nameof(expectedVersion));
+
+            return AppendEventsInternalAsync(commitId, streamName, expectedVersion, events);
+        }
+
+        private async Task AppendEventsInternalAsync(Guid commitId, string streamName, long expectedVersion, ICollection<EventData> events)
         {
             Guard.NotNullOrEmpty(streamName, nameof(streamName));
             Guard.NotNull(events, nameof(events));
 
-            var eventsCount = events.Count;
+            if (events.Count == 0)
-
-            if (eventsCount > 0)
             {
-                var commitEvents = new MongoEvent[events.Count];
+                return;
+            }
 
-                var i = 0;
+            var currentVersion = await GetEventStreamOffset(streamName);
 
-                foreach (var e in events)
+            if (expectedVersion != AnyVersion && expectedVersion != currentVersion)
-                {
+            {
-                    var mongoEvent = new MongoEvent { EventId = e.EventId, Metadata = e.Metadata, Payload = e.Payload, Type = e.Type };
+                throw new WrongEventVersionException(currentVersion, expectedVersion);
+            }
 
-                    commitEvents[i++] = mongoEvent;
+            var commit = BuildCommit(commitId, streamName, expectedVersion >= -1 ? expectedVersion : currentVersion, events);
-                }
 
+            for (var attempt = 0; attempt < MaxAttempts; attempt++)
+            {
                 try
                 {
-                    var document = new MongoEventCommit
+                    await Collection.InsertOneAsync(commit);
-                    {
-                        Id = commitId,
-                        Events = commitEvents,
-                        EventsCount = eventsCount,
-                        EventStream = streamName,
-                        EventStreamOffset = expectedVersion,
-                        Timestamp = EmptyTimestamp
-                    };
-
-                    await Collection.InsertOneAsync(document);
 
                     notifier.NotifyEventsStored();
+
+                    return;
                 }
                 catch (MongoWriteException ex)
                 {
                     if (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
                     {
-                        var currentVersion = await GetEventStreamOffset(streamName);
+                        currentVersion = await GetEventStreamOffset(streamName);
-
+
-                        throw new WrongEventVersionException(currentVersion, expectedVersion);
+                        if (expectedVersion != AnyVersion)
+                        {
+                            throw new WrongEventVersionException(currentVersion, expectedVersion);
+                        }
+                        else if (attempt < MaxAttempts)
+                        {
+                            expectedVersion = currentVersion;
+                        }
+                        else
+                        {
+                            throw new TimeoutException("Could not acquire a free slot for the commit within the provided time.");
+                        }
                     }
                     else
                     {
@@ -204,5 +223,31 @@ namespace Squidex.Infrastructure.CQRS.Events
 
             return Filter.And(filters);
         }
+
+        private static MongoEventCommit BuildCommit(Guid commitId, string streamName, long expectedVersion, ICollection<EventData> events)
+        {
+            var commitEvents = new MongoEvent[events.Count];
+
+            var i = 0;
+
+            foreach (var e in events)
+            {
+                var mongoEvent = new MongoEvent { EventId = e.EventId, Metadata = e.Metadata, Payload = e.Payload, Type = e.Type };
+
+                commitEvents[i++] = mongoEvent;
+            }
+
+            var mongoCommit = new MongoEventCommit
+            {
+                Id = commitId,
+                Events = commitEvents,
+                EventsCount = events.Count,
+                EventStream = streamName,
+                EventStreamOffset = expectedVersion,
+                Timestamp = EmptyTimestamp
+            };
+
+            return mongoCommit;
+        }
     }
 }

src/Squidex.Infrastructure/CQRS/Events/IEventStore.cs

@@ -16,6 +16,8 @@ namespace Squidex.Infrastructure.CQRS.Events
     {
         Task<IReadOnlyList<StoredEvent>> GetEventsAsync(string streamName);
 
+        Task AppendEventsAsync(Guid commitId, string streamName, ICollection<EventData> events);
+
         Task AppendEventsAsync(Guid commitId, string streamName, int expectedVersion, ICollection<EventData> events);
 
         IEventSubscription CreateSubscription(string streamFilter = null, string position = null);

tests/Benchmarks/Tests/AppendToEventStoreWithManyWriters.cs

@@ -51,13 +51,11 @@ namespace Benchmarks.Tests
 
             Parallel.For(0, numStreams, streamId =>
             {
-                var eventOffset = -1;
                 var streamName = streamId.ToString();
 
                 for (var commitId = 0; commitId < numCommits; commitId++)
                 {
-                    eventStore.AppendEventsAsync(Guid.NewGuid(), streamName, eventOffset, new[] { Helper.CreateEventData() }).Wait();
+                    eventStore.AppendEventsAsync(Guid.NewGuid(), streamName, new[] { Helper.CreateEventData() }).Wait();
-                    eventOffset++;
                 }
             });
 

tests/Benchmarks/Utils/Helper.cs

@@ -21,7 +21,7 @@ namespace Benchmarks.Utils
 
         public static void Warmup(this IEventStore eventStore)
         {
-            eventStore.AppendEventsAsync(Guid.NewGuid(), "my-stream", -1, new List<EventData> { CreateEventData() }).Wait();
+            eventStore.AppendEventsAsync(Guid.NewGuid(), "my-stream", new List<EventData> { CreateEventData() }).Wait();
         }
     }
 }

tests/Squidex.Infrastructure.Tests/CQRS/Events/EventReceiverTests.cs

@@ -88,6 +88,11 @@ namespace Squidex.Infrastructure.CQRS.Events
                 throw new NotSupportedException();
             }
 
+            public Task AppendEventsAsync(Guid commitId, string streamName, ICollection<EventData> events)
+            {
+                throw new NotSupportedException();
+            }
+
             public Task AppendEventsAsync(Guid commitId, string streamName, int expectedVersion, ICollection<EventData> events)
             {
                 throw new NotSupportedException();