add integration tests for claimMinIdleTime

Author: mgravell

src/StackExchange.Redis/RedisFeatures.cs

@@ -46,7 +46,8 @@ namespace StackExchange.Redis
                                          v7_4_0_rc1 = new Version(7, 3, 240), // 7.4 RC1 is version 7.3.240
                                          v7_4_0_rc2 = new Version(7, 3, 241), // 7.4 RC2 is version 7.3.241
                                          v8_0_0_M04 = new Version(7, 9, 227), // 8.0 M04 is version 7.9.227
-                                         v8_2_0_rc1 = new Version(8, 1, 240); // 8.2 RC1 is version 8.1.240
+                                         v8_2_0_rc1 = new Version(8, 1, 240), // 8.2 RC1 is version 8.1.240
+                                         v8_4_0_rc1 = new Version(8, 3, 224); // 8.2 RC1 is version 8.3.224
 
 #pragma warning restore SA1310 // Field names should not contain underscore
 #pragma warning restore SA1311 // Static readonly fields should begin with upper-case letter

tests/StackExchange.Redis.Tests/StreamTests.cs

@@ -495,8 +495,8 @@ public async Task StreamConsumerGroupSetId()
 
         var db = conn.GetDatabase();
         var key = Me();
-        const string groupName = "test_group",
-                     consumer = "consumer";
+        await db.KeyDeleteAsync(key, CommandFlags.FireAndForget);
+        const string groupName = "test_group", consumer = "consumer";
 
         // Create a stream
         db.StreamAdd(key, "field1", "value1");
@@ -519,6 +519,56 @@ public async Task StreamConsumerGroupSetId()
         Assert.Equal(2, secondRead.Length);
     }
 
+    [Fact]
+    public async Task StreamConsumerGroupAutoClaim()
+    {
+        await using var conn = Create(require: RedisFeatures.v8_4_0_rc1);
+
+        var db = conn.GetDatabase();
+        var key = Me();
+        await db.KeyDeleteAsync(key, CommandFlags.FireAndForget);
+        const string groupName = "test_group", consumer = "consumer";
+
+        // Create a group and set the position to deliver new messages only.
+        await db.StreamCreateConsumerGroupAsync(key, groupName, StreamPosition.NewMessages);
+
+        // add some entries
+        await db.StreamAddAsync(key, "field1", "value1");
+        await db.StreamAddAsync(key, "field2", "value2");
+
+        var idleTime = TimeSpan.FromMilliseconds(100);
+        // Read into the group, expect the two entries; we don't expect any data
+        // here, at least on a fast server, because it hasn't been idle long enough.
+        StreamPosition[] positions = [new(key, StreamPosition.NewMessages)];
+        var groups = await db.StreamReadGroupAsync(positions, groupName, consumer, noAck: false, countPerStream: 10, claimMinIdleTime: idleTime);
+        var grp = Assert.Single(groups);
+        Assert.Equal(key, grp.Key);
+        Assert.Equal(2, grp.Entries.Length);
+        foreach (var entry in grp.Entries)
+        {
+            Assert.Equal(0, entry.DeliveryCount); // never delivered before
+            Assert.Equal(TimeSpan.Zero, entry.IdleTime); // never delivered before
+        }
+
+        // now repeat immediately; we didn't "ack", so they're still pending, but not idle long enough
+        groups = await db.StreamReadGroupAsync(positions, groupName, consumer, noAck: false, countPerStream: 10, claimMinIdleTime: idleTime);
+        Assert.Empty(groups); // nothing available from any group
+
+        // wait long enough for the messages to be considered idle
+        await Task.Delay(idleTime + idleTime);
+
+        // repeat again; we should get the entries
+        groups = await db.StreamReadGroupAsync(positions, groupName, consumer, noAck: false, countPerStream: 10, claimMinIdleTime: idleTime);
+        grp = Assert.Single(groups);
+        Assert.Equal(key, grp.Key);
+        Assert.Equal(2, grp.Entries.Length);
+        foreach (var entry in grp.Entries)
+        {
+            Assert.Equal(1, entry.DeliveryCount); // this is a redelivery
+            Assert.True(entry.IdleTime > TimeSpan.Zero); // and is considered idle
+        }
+    }
+
     [Fact]
     public async Task StreamConsumerGroupWithNoConsumers()
     {