T

events/Squidex.Events.GetEventStore/GetEventStore.cs

@@ -72,9 +72,14 @@ public async IAsyncEnumerable<StoredEvent> QueryAllReverseAsync(StreamFilter fil
         }
 
         var streamName = await projectionClient.CreateProjectionAsync(filter, true, ct);
-        var streamEvents = QueryReverseAsync(streamName, ESStreamPosition.End, take, ct);
+        var streamEvents = QueryReverseAsync(streamName, ESStreamPosition.End, int.MaxValue, ct);
 
-        await foreach (var storedEvent in streamEvents.IgnoreNotFound(ct).TakeWhile(x => x.Data.Headers.Timestamp() >= timestamp).WithCancellation(ct))
+        var query = streamEvents
+            .IgnoreNotFound(ct)
+            .TakeWhile(x => x.Data.Headers.Timestamp() >= timestamp)
+            .Take(take);
+
+        await foreach (var storedEvent in query.WithCancellation(ct))
         {
             yield return storedEvent;
         }

events/Squidex.Events.Tests/EnvelopeHeadersTests.cs

@@ -5,6 +5,7 @@
 //  All rights reserved. Licensed under the MIT license.
 // ==========================================================================
 
+using FakeItEasy;
 using MongoDB.Bson.Serialization;
 using Squidex.Events.Mongo;
 using Xunit;
@@ -18,6 +19,138 @@ static EnvelopeHeadersTests()
         BsonSerializer.TryRegisterSerializer(new HeaderValueSerializer());
     }
 
+    [Fact]
+    public void Should_get_long()
+    {
+        var headers = new EnvelopeHeaders
+        {
+            ["long"] = 42
+        };
+
+        var result = headers.GetLong("long");
+        Assert.Equal(42, result);
+    }
+
+    [Fact]
+    public void Should_get_long_from_empty_key()
+    {
+        var headers = new EnvelopeHeaders
+        {
+        };
+
+        var result = headers.GetLong("long");
+        Assert.Equal(0, result);
+    }
+
+    [Theory]
+    [InlineData("9", 9)]
+    [InlineData("A", 0)]
+    [InlineData(" ", 0)]
+    public void Should_get_long_from_string(string source, long expected)
+    {
+        var headers = new EnvelopeHeaders
+        {
+            ["long"] = source
+        };
+
+        var result = headers.GetLong("long");
+        Assert.Equal(expected, result);
+    }
+
+    [Fact]
+    public void Should_get_string()
+    {
+        var headers = new EnvelopeHeaders
+        {
+            ["string"] = "Hello"
+        };
+
+        var result = headers.GetString("string");
+        Assert.Equal("Hello", result);
+    }
+
+    [Fact]
+    public void Should_get_string_from_empty_key()
+    {
+        var headers = new EnvelopeHeaders
+        {
+        };
+
+        var result = headers.GetString("string");
+        Assert.Equal(string.Empty, result);
+    }
+
+    [Fact]
+    public void Should_get_string_from_long()
+    {
+        var headers = new EnvelopeHeaders
+        {
+            ["string"] = 42
+        };
+
+        var result = headers.GetString("string");
+        Assert.Equal("42", result);
+    }
+
+    [Fact]
+    public void Should_get_string_from_bool()
+    {
+        var headers = new EnvelopeHeaders
+        {
+            ["string"] = true
+        };
+
+        var result = headers.GetString("string");
+        Assert.Equal("true", result);
+    }
+
+    [Fact]
+    public void Should_get_boolean()
+    {
+        var headers = new EnvelopeHeaders
+        {
+            ["bool"] = true
+        };
+
+        var result = headers.GetBoolean("bool");
+        Assert.True(result);
+    }
+
+    [Fact]
+    public void Should_get_boolean_from_empty_key()
+    {
+        var headers = new EnvelopeHeaders
+        {
+        };
+
+        var result = headers.GetBoolean("bool");
+        Assert.False(result);
+    }
+
+    [Fact]
+    public void Should_get_datetime()
+    {
+        var headers = new EnvelopeHeaders
+        {
+            ["date"] = "2023-12-11T10:09:08z"
+        };
+
+        var result = headers.GetDateTime("date");
+        Assert.Equal(new DateTime(2023, 12, 11, 10, 9, 8, DateTimeKind.Utc), result);
+    }
+
+    [Fact]
+    public void Should_get_datetime_with_millis()
+    {
+        var headers = new EnvelopeHeaders
+        {
+            ["date"] = "2023-12-11T10:09:08.765z"
+        };
+
+        var result = headers.GetDateTime("date");
+        Assert.Equal(new DateTime(2023, 12, 11, 10, 9, 8, 765, DateTimeKind.Utc), result);
+    }
+
     [Fact]
     public void Should_create_headers()
     {

events/Squidex.Events.Tests/EventStoreTests.cs

@@ -263,8 +263,8 @@ public async Task Should_subscribe_with_parallel_writes()
         var expectedEvents = numTasks * numEvents;
 
         // Append and read in parallel.
-        //var readEvents = await QueryWithSubscriptionAsync(sut, streamFilter, expectedEvents, async () =>
-        // {
+        var readEvents = await QueryWithSubscriptionAsync(sut, streamFilter, expectedEvents, async () =>
+        {
             await Parallel.ForEachAsync(Enumerable.Range(0, numTasks), async (i, ct) =>
             {
                 var fullStreamName = $"{streamName}-{Guid.NewGuid()}";
@@ -279,9 +279,9 @@ await Parallel.ForEachAsync(Enumerable.Range(0, numTasks), async (i, ct) =>
                     await sut.AppendAsync(Guid.NewGuid(), fullStreamName, EtagVersion.Any, commit);
                 }
             });
-        // });
+        });
 
-        // Assert.Equal(expectedEvents, readEvents?.Count);
+        Assert.Equal(expectedEvents, readEvents?.Count);
     }
 
     [Fact]
@@ -510,7 +510,7 @@ private static EventData CreateEventData(int i)
     {
         var headers = new EnvelopeHeaders
         {
-            [CommonHeaders.EventId] = Guid.NewGuid().ToString()
+            ["EventId"] = Guid.NewGuid().ToString()
         };
 
         return new EventData($"Type{i}", headers, i.ToString(CultureInfo.InvariantCulture));

events/Squidex.Events.Tests/PollingSubscriptionTests.cs

@@ -131,7 +131,7 @@ private StoredEvent CreateEvent(int offset)
                 "type",
                 new EnvelopeHeaders
                 {
-                    [CommonHeaders.EventId] = Guid.NewGuid().ToString()
+                    ["EventId"] = Guid.NewGuid().ToString()
                 },
                 "payload"));
     }

events/Squidex.Events/CoreHeaders.cs

@@ -0,0 +1,13 @@
+// ==========================================================================
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex UG (haftungsbeschraenkt)
+//  All rights reserved. Licensed under the MIT license.
+// ==========================================================================
+
+namespace Squidex.Events;
+
+public static class CoreHeaders
+{
+    public static readonly string Timestamp = nameof(Timestamp);
+}

events/Squidex.Events/EnvelopeExtensions.cs

@@ -11,6 +11,24 @@ namespace Squidex.Events;
 
 public static class EnvelopeExtensions
 {
+    public static DateTime Timestamp(this EnvelopeHeaders obj)
+    {
+        return obj.GetDateTime(CoreHeaders.Timestamp);
+    }
+
+    public static DateTime GetDateTime(this EnvelopeHeaders obj, string key)
+    {
+        if (obj.TryGetValue(key, out var found))
+        {
+            if (found is HeaderStringValue s && DateTime.TryParse(s.Value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out var dateTime))
+            {
+                return dateTime;
+            }
+        }
+
+        return default;
+    }
+
     public static long GetLong(this EnvelopeHeaders obj, string key)
     {
         if (obj.TryGetValue(key, out var found))
@@ -19,7 +37,8 @@ public static long GetLong(this EnvelopeHeaders obj, string key)
             {
                 return n.Value;
             }
-            else if (found is HeaderStringValue s && double.TryParse(s.Value, NumberStyles.Any, CultureInfo.InvariantCulture, out var result))
+
+            if (found is HeaderStringValue s && double.TryParse(s.Value, NumberStyles.Any, CultureInfo.InvariantCulture, out var result))
             {
                 return (long)result;
             }

events/Squidex.Events/EventHeaderValue.cs

@@ -32,7 +32,7 @@ public record HeaderBooleanValue(bool Value) : HeaderValue
 {
     public override string ToString()
     {
-        return Value.ToString();
+        return Value ? "true" : "false";
     }
 }