From 5c56c7ca1a5e712fb1f7fcead655782acd123e7f Mon Sep 17 00:00:00 2001
From: Thorsten Thiel <mail@thorsten-thiel.com>
Date: Fri, 6 Sep 2024 23:26:17 +0200
Subject: [PATCH] WIP

---
 src/Benchmark/Bench.cs                           |  4 +++-
 src/Benchmark/Program.cs                         | 11 ++++++++++-
 src/Fluss/UnitOfWork/IUnitOfWork.cs              |  2 +-
 src/Fluss/UnitOfWork/UnitOfWork.Aggregates.cs    |  8 +++++---
 src/Fluss/UnitOfWork/UnitOfWorkRecordingProxy.cs |  2 +-
 src/Fluss/Validation/RootValidator.cs            |  2 +-
 6 files changed, 21 insertions(+), 8 deletions(-)

diff --git a/src/Benchmark/Bench.cs b/src/Benchmark/Bench.cs
index 39ef892..e4a9ba0 100644
--- a/src/Benchmark/Bench.cs
+++ b/src/Benchmark/Bench.cs
@@ -1,4 +1,5 @@
-using BenchmarkDotNet.Attributes;
+using BenchmarkDotNet_GitCompare;
+using BenchmarkDotNet.Attributes;
 using Fluss;
 using Fluss.Authentication;
 using Fluss.Events;
@@ -8,6 +9,7 @@
 namespace Benchmark;
 
 [SimpleJob]
+[GitJob(baseline: true)]
 [MemoryDiagnoser]
 public class Bench
 {
diff --git a/src/Benchmark/Program.cs b/src/Benchmark/Program.cs
index d7bd483..6f77bff 100644
--- a/src/Benchmark/Program.cs
+++ b/src/Benchmark/Program.cs
@@ -1,4 +1,13 @@
 using Benchmark;
 using BenchmarkDotNet.Running;
 
-BenchmarkRunner.Run<Bench>();
\ No newline at end of file
+if (false)
+{
+    BenchmarkRunner.Run<Bench>();
+}
+else
+{
+    var bench = new Bench();
+    bench.Setup();
+    await bench.PublishEventsAndRead();
+}
diff --git a/src/Fluss/UnitOfWork/IUnitOfWork.cs b/src/Fluss/UnitOfWork/IUnitOfWork.cs
index abfb966..c3d47d9 100644
--- a/src/Fluss/UnitOfWork/IUnitOfWork.cs
+++ b/src/Fluss/UnitOfWork/IUnitOfWork.cs
@@ -8,7 +8,7 @@ public interface IUnitOfWork
 {
     ValueTask<long> ConsistentVersion();
     IReadOnlyCollection<EventListener> ReadModels { get; }
-    ConcurrentQueue<EventEnvelope> PublishedEventEnvelopes { get; }
+    IReadOnlyCollection<EventEnvelope> PublishedEventEnvelopes { get; }
 
     ValueTask<IReadModel> GetReadModel(Type tReadModel, object? key, long? at = null);
 
diff --git a/src/Fluss/UnitOfWork/UnitOfWork.Aggregates.cs b/src/Fluss/UnitOfWork/UnitOfWork.Aggregates.cs
index 6ea222a..ce82bab 100644
--- a/src/Fluss/UnitOfWork/UnitOfWork.Aggregates.cs
+++ b/src/Fluss/UnitOfWork/UnitOfWork.Aggregates.cs
@@ -7,7 +7,9 @@ namespace Fluss;
 
 public partial class UnitOfWork
 {
-    public ConcurrentQueue<EventEnvelope> PublishedEventEnvelopes { get; } = new();
+    // TODO: Use https://github.com/jtmueller/Collections.Pooled
+    internal readonly List<EventEnvelope> _publishedEventEnvelopes = [];
+    public IReadOnlyCollection<EventEnvelope> PublishedEventEnvelopes => _publishedEventEnvelopes;
 
     public async ValueTask<TAggregate> GetAggregate<TAggregate>() where TAggregate : AggregateRoot, new()
     {
@@ -68,7 +70,7 @@ public async ValueTask Publish(Event @event, AggregateRoot? aggregate = null)
 
         await ValidateEventResult(eventEnvelope, aggregate);
 
-        PublishedEventEnvelopes.Enqueue(eventEnvelope);
+        _publishedEventEnvelopes.Add(eventEnvelope);
     }
 
     private async ValueTask ValidateEventResult<T>(EventEnvelope envelope, T? aggregate) where T : AggregateRoot
@@ -96,7 +98,7 @@ internal async ValueTask CommitInternal()
 
         await _eventRepository.Publish(PublishedEventEnvelopes);
         _consistentVersion += PublishedEventEnvelopes.Count;
-        PublishedEventEnvelopes.Clear();
+        _publishedEventEnvelopes.Clear();
     }
 
     private async ValueTask<TEventListener> UpdateAndApplyPublished<TEventListener>(TEventListener eventListener, long? at)
diff --git a/src/Fluss/UnitOfWork/UnitOfWorkRecordingProxy.cs b/src/Fluss/UnitOfWork/UnitOfWorkRecordingProxy.cs
index b5a9001..4c4cb5b 100644
--- a/src/Fluss/UnitOfWork/UnitOfWorkRecordingProxy.cs
+++ b/src/Fluss/UnitOfWork/UnitOfWorkRecordingProxy.cs
@@ -12,7 +12,7 @@ public ValueTask<long> ConsistentVersion()
     }
 
     public IReadOnlyCollection<EventListener> ReadModels => impl.ReadModels;
-    public ConcurrentQueue<EventEnvelope> PublishedEventEnvelopes => impl.PublishedEventEnvelopes;
+    public IReadOnlyCollection<EventEnvelope> PublishedEventEnvelopes => impl.PublishedEventEnvelopes;
 
     public List<EventListener> RecordedListeners { get; } = [];
 
diff --git a/src/Fluss/Validation/RootValidator.cs b/src/Fluss/Validation/RootValidator.cs
index 7fe8757..f3c543c 100644
--- a/src/Fluss/Validation/RootValidator.cs
+++ b/src/Fluss/Validation/RootValidator.cs
@@ -65,7 +65,7 @@ public async Task ValidateEvent(EventEnvelope envelope, IReadOnlyList<EventEnvel
         var versionedUnitOfWork = unitOfWork.WithPrefilledVersion(envelope.Version - willBePublishedEnvelopes.Count - 1);
         foreach (var willBePublishedEnvelope in willBePublishedEnvelopes)
         {
-            versionedUnitOfWork.PublishedEventEnvelopes.Enqueue(willBePublishedEnvelope);
+            ((List<EventEnvelope>)versionedUnitOfWork.PublishedEventEnvelopes).Add(willBePublishedEnvelope);
         }
 
         var type = envelope.Event.GetType();