refactor: update InMemoryStream && add InMemoryStreamTest

WeihanLi · Dec 15, 2024 · c243d8e · c243d8e
1 parent d0f0267
commit c243d8e
Show file tree

Hide file tree

Showing 3 changed files with 56 additions and 3 deletions.
diff --git a/samples/DotNetCoreSample/InMemoryStreamTest.cs b/samples/DotNetCoreSample/InMemoryStreamTest.cs
@@ -0,0 +1,44 @@
+// Copyright (c) Weihan Li. All rights reserved.
+// Licensed under the Apache license.
+
+using WeihanLi.Common.Helpers;
+using WeihanLi.Extensions;
+
+namespace DotNetCoreSample;
+internal static class MemoryStreamTest
+{
+    public static async Task MainTest()
+    {
+        var stream = new InMemoryStream<long>("test");
+        var timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+        await Task.Delay(100);
+        await stream.AddAsync(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), new Dictionary<string, string>
+        {
+            { "messages", new { name = $"test-{DateTimeOffset.Now}" } .ToJson() }
+        });
+        Console.WriteLine("stream message added");
+        await Task.Delay(1000);
+        await stream.AddAsync(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), new Dictionary<string, string>
+        {
+            { "messages", new { name = $"test-{DateTimeOffset.Now}" } .ToJson() }
+        });
+        Console.WriteLine("stream message added");
+        //
+        {
+            Console.WriteLine("Fetch messages from stream");
+            await foreach (var item in stream.FetchAsync(timestamp, 2))
+            {
+                Console.WriteLine($"{item.Id} - {item.Timestamp}");
+                Console.WriteLine(item.Fields.ToJson());
+            }
+        }
+        {
+            Console.WriteLine("Fetch messages from stream again");
+            await foreach (var item in stream.FetchAsync(timestamp, 2))
+            {
+                Console.WriteLine($"{item.Id} - {item.Timestamp}");
+                Console.WriteLine(item.Fields.ToJson());
+            }
+        }
+    }
+}
diff --git a/samples/DotNetCoreSample/Program.cs b/samples/DotNetCoreSample/Program.cs
@@ -347,7 +347,8 @@
 
 // InvokeHelper.TryInvoke(() => throw null, 3);
 
-InvokeHelper.TryInvoke(LoggerTest.MicrosoftLoggingTest);
+// InvokeHelper.TryInvoke(LoggerTest.MicrosoftLoggingTest);
+await InvokeHelper.TryInvokeAsync(MemoryStreamTest.MainTest);
 
 ConsoleHelper.ReadKeyWithPrompt("Press any key to exit");
 

diff --git a/src/WeihanLi.Common/Helpers/InMemoryStream.cs b/src/WeihanLi.Common/Helpers/InMemoryStream.cs
@@ -73,9 +73,17 @@ public Task AddAsync(T id, Dictionary<string, string> fields, DateTimeOffset? ti
         {
             Id = id,
             Fields = fields,
-            Timestamp = timestamp ?? DateTimeOffset.Now,
-            Properties = properties ?? new()
+            Timestamp = timestamp ?? DateTimeOffset.Now
         };
+
+        if (properties is { Count: > 0 })
+        {
+            foreach (var item in properties)
+            {
+                message.Properties[item.Key] = item.Value;
+            }
+        }
+
         _messages.Add(message);
 
         return Task.CompletedTask;