Redis queue: round-trip GroupId

src/Foundatio.Redis/Queues/RedisQueue.cs

@@ -218,6 +218,7 @@
         {
             Properties = options.Properties,
             CorrelationId = options.CorrelationId,
+            GroupId = options.GroupId,
             Value = data
         };
         bool success = await _resiliencePolicy.ExecuteAsync(async _ => await _cache.AddAsync(GetPayloadKey(id), envelope, _payloadTimeToLive)).AnyContext();
@@ -240,7 +241,7 @@
         }
 
         Interlocked.Increment(ref _enqueuedCount);
-        var entry = new QueueEntry<T>(id, options.CorrelationId, data, this, now, 0);
+        var entry = new QueueEntry<T>(id, options.CorrelationId, data, this, now, 0) { GroupId = options.GroupId };
         await OnEnqueuedAsync(entry).AnyContext();
 
         _logger.LogTrace("Enqueue done");
@@ -406,7 +407,7 @@
         var attemptsValue = _resiliencePolicy.ExecuteAsync(async _ => await _cache.GetAsync(GetAttemptsKey(workId), 0)).AsTask();
         await Task.WhenAll(enqueuedTimeTicks, attemptsValue).AnyContext();
 
-        var queueEntry = new QueueEntry<T>(workId, payload.Value.CorrelationId, payload.Value.Value, this, new DateTime(enqueuedTimeTicks.Result, DateTimeKind.Utc), attemptsValue.Result + 1);
+        var queueEntry = new QueueEntry<T>(workId, payload.Value.CorrelationId, payload.Value.Value, this, new DateTime(enqueuedTimeTicks.Result, DateTimeKind.Utc), attemptsValue.Result + 1) { GroupId = payload.Value.GroupId };
 
         if (payload.Value.Properties != null)
         {
@@ -894,6 +895,7 @@
 public record RedisPayloadEnvelope<T>
 {
     public string? CorrelationId { get; init; }
+    public string? GroupId { get; init; }
     public IDictionary<string, string>? Properties { get; init; }
     public required T Value { get; init; }
 }

tests/Foundatio.Redis.Tests/Queues/RedisQueueTests.cs

@@ -120,6 +120,12 @@ public override Task DequeueAsync_AfterAbandonWithMutatedValue_ReturnsOriginalVa
         return base.DequeueAsync_AfterAbandonWithMutatedValue_ReturnsOriginalValueAsync();
     }
 
+    [Fact]
+    public override Task AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync()
+    {
+        return base.AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync();
+    }
+
     [Fact]
     public override Task DequeueAsync_WithDispose_AutoAbandonsEntryAsync()
     {
@@ -162,6 +168,12 @@ public override Task DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync(
         return base.DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync();
     }
 
+    [Fact]
+    public override Task EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync()
+    {
+        return base.EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync();
+    }
+
     [Fact]
     public override Task EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync()
     {