Tests

Squidex · Jan 15, 2025 · 6861093 · 6861093
1 parent 74e315f
commit 6861093
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 13 deletions.
diff --git a/events/Squidex.Events.EntityFramework/EFEventStore_Writer.cs b/events/Squidex.Events.EntityFramework/EFEventStore_Writer.cs
@@ -59,8 +59,18 @@ public async Task AppendAsync(Guid commitId, string streamName, long expectedVer
 
                 try
                 {
-                    commit.Position = await adapter.GetPositionAsync(context, ct);
-                    await context.SaveChangesAsync(ct);
+                    await using var transaction = await context.Database.BeginTransactionAsync(ct);
+                    try
+                    {
+                        commit.Position = await adapter.GetPositionAsync(context, ct);
+                        await context.SaveChangesAsync(ct);
+                        await transaction.CommitAsync(ct);
+                    }
+                    catch (Exception)
+                    {
+                        await transaction.RollbackAsync(ct);
+                        throw;
+                    }
                 }
                 catch
                 {

diff --git a/events/Squidex.Events.GetEventStore/GetEventStore.cs b/events/Squidex.Events.GetEventStore/GetEventStore.cs
@@ -72,7 +72,7 @@ public async IAsyncEnumerable<StoredEvent> QueryAllReverseAsync(StreamFilter fil
         }
 
         var streamName = await projectionClient.CreateProjectionAsync(filter, true, ct);
-        var streamEvents = QueryReverseAsync(streamName, ESStreamPosition.End, int.MaxValue, ct);
+        var streamEvents = QueryReverseAsync(streamName, take, ct);
 
         var query = streamEvents
             .IgnoreNotFound(ct)
@@ -115,13 +115,13 @@ private IAsyncEnumerable<StoredEvent> QueryAsync(string streamName, ESStreamPosi
         return result.Select(x => Formatter.Read(x, options.Value.Prefix));
     }
 
-    private IAsyncEnumerable<StoredEvent> QueryReverseAsync(string streamName, ESStreamPosition start, long count,
+    private IAsyncEnumerable<StoredEvent> QueryReverseAsync(string streamName, long count,
         CancellationToken ct = default)
     {
         var result = client.ReadStreamAsync(
             Direction.Backwards,
             streamName,
-            start,
+            ESStreamPosition.End,
             count,
             true,
             cancellationToken: ct);

diff --git a/events/Squidex.Events.Tests/EventStoreTests.cs b/events/Squidex.Events.Tests/EventStoreTests.cs
@@ -516,7 +516,7 @@ private static EventData CreateEventData(int i)
         return new EventData($"Type{i}", headers, i.ToString(CultureInfo.InvariantCulture));
     }
 
-    private async Task<IReadOnlyList<StoredEvent>?> QueryWithSubscriptionAsync(
+    private async Task<IReadOnlyList<StoredEvent>> QueryWithSubscriptionAsync(
         IEventStore sut,
         StreamFilter streamFilter,
         int expectedCount,
@@ -546,14 +546,16 @@ private static EventData CreateEventData(int i)
                     if (subscriber.LastEvents.Count >= expectedCount)
                     {
                         subscriptionPosition = subscriber.LastPosition;
-
                         return subscriber.LastEvents;
                     }
                 }
-
-                cts.Token.ThrowIfCancellationRequested();
-                return null;
             }
+
+            return subscriber.LastEvents;
+        }
+        catch (OperationCanceledException)
+        {
+            return subscriber.LastEvents;
         }
         finally
         {

diff --git a/events/Squidex.Events/PollingSubscription.cs b/events/Squidex.Events/PollingSubscription.cs
@@ -12,6 +12,9 @@ namespace Squidex.Events;
 public sealed class PollingSubscription : IEventSubscription
 {
     private readonly CompletionTimer timer;
+#pragma warning disable IDE0052 // Remove unread private members
+    private int eventsTotal;
+#pragma warning restore IDE0052 // Remove unread private members
 
     public PollingSubscription(
         IEventStore eventStore,
@@ -29,16 +32,17 @@ public PollingSubscription(
             {
                 while (true)
                 {
-                    var hasAddedEvent = false;
+                    var eventsInAttempt = 0;
                     await foreach (var storedEvent in eventStore.QueryAllAsync(streamFilter, streamPosition, ct: ct))
                     {
                         await eventSubscriber.OnNextAsync(this, storedEvent);
 
                         streamPosition = storedEvent.EventPosition;
-                        hasAddedEvent = true;
+                        eventsInAttempt++;
+                        eventsTotal++;
                     }
 
-                    if (!hasAddedEvent)
+                    if (eventsInAttempt == 0)
                     {
                         break;
                     }