StreamGroupInfo.Lag can be null (#2902)

* TryRead for nullable long, tests on null value for StreamConsumerGroupInfo.Lag

* Update StreamTests.cs

make tests async

* Update StreamTests.cs

tyop

---------

Co-authored-by: Marc Gravell <marc.gravell@gmail.com>

Author: robhop

src/StackExchange.Redis/ResultProcessor.cs

@@ -2289,6 +2289,19 @@ internal static bool TryRead(Sequence<RawResult> pairs, in CommandBytes key, ref
                 }
                 return false;
             }
+            internal static bool TryRead(Sequence<RawResult> pairs, in CommandBytes key, ref long? value)
+            {
+                var len = pairs.Length / 2;
+                for (int i = 0; i < len; i++)
+                {
+                    if (pairs[i * 2].IsEqual(key) && pairs[(i * 2) + 1].TryGetInt64(out var tmp))
+                    {
+                        value = tmp;
+                        return true;
+                    }
+                }
+                return false;
+            }
 
             internal static bool TryRead(Sequence<RawResult> pairs, in CommandBytes key, ref int value)
             {
@@ -2351,7 +2364,8 @@ protected override StreamGroupInfo ParseItem(in RawResult result)
                 var arr = result.GetItems();
                 string? name = default, lastDeliveredId = default;
                 int consumerCount = default, pendingMessageCount = default;
-                long entriesRead = default, lag = default;
+                long entriesRead = default;
+                long? lag = default;
 
                 KeyValuePairParser.TryRead(arr, KeyValuePairParser.Name, ref name);
                 KeyValuePairParser.TryRead(arr, KeyValuePairParser.Consumers, ref consumerCount);

tests/StackExchange.Redis.Tests/StreamTests.cs

@@ -2036,4 +2036,43 @@ await db.StreamAddAsync(
         Assert.Equal(123, (int)obj!.id);
         Assert.Equal("test", (string)obj.name);
     }
+
+    [Fact]
+    public async Task StreamConsumerGroupInfoLagIsNull()
+    {
+        await using var conn = Create(require: RedisFeatures.v5_0_0);
+
+        var db = conn.GetDatabase();
+        var key = Me();
+        const string groupName = "test_group",
+                     consumer = "consumer";
+
+        await db.StreamCreateConsumerGroupAsync(key, groupName);
+        await db.StreamReadGroupAsync(key, groupName, consumer, "0-0", 1);
+        await db.StreamAddAsync(key, "field1", "value1");
+        await db.StreamAddAsync(key, "field1", "value1");
+
+        var streamInfo = await db.StreamInfoAsync(key);
+        await db.StreamDeleteAsync(key, new[] { streamInfo.LastEntry.Id });
+
+        Assert.Null((await db.StreamGroupInfoAsync(key))[0].Lag);
+    }
+
+    [Fact]
+    public async Task StreamConsumerGroupInfoLagIsTwo()
+    {
+        await using var conn = Create(require: RedisFeatures.v5_0_0);
+
+        var db = conn.GetDatabase();
+        var key = Me();
+        const string groupName = "test_group",
+                     consumer = "consumer";
+
+        await db.StreamCreateConsumerGroupAsync(key, groupName);
+        await db.StreamReadGroupAsync(key, groupName, consumer, "0-0", 1);
+        await db.StreamAddAsync(key, "field1", "value1");
+        await db.StreamAddAsync(key, "field1", "value1");
+
+        Assert.Equal(2, (await db.StreamGroupInfoAsync(key))[0].Lag);
+    }
 }