WIP

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
 {

src/Benchmark/Program.cs

@@ -1,4 +1,13 @@
 using Benchmark;
 using BenchmarkDotNet.Running;
 
-BenchmarkRunner.Run<Bench>();
+if (false)
+{
+    BenchmarkRunner.Run<Bench>();
+}
+else
+{
+    var bench = new Bench();
+    bench.Setup();
+    await bench.PublishEventsAndRead();
+}

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);
 

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)

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; } = [];
 

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();