Add Prioritized Priority Consumer Policy

src/NATS.Client.JetStream/Models/ConsumerConfig.cs

@@ -272,7 +272,7 @@ public ConsumerConfig(string name)
 #endif
 
     /// <summary>
-    /// Specifies the priority policy for consumer message selection, such as prioritizing <c>overflow</c> or <c>pinned_client</c>.
+    /// Specifies the priority policy for consumer message selection, such as prioritizing <c>prioritized</c>, <c>overflow</c>, or <c>pinned_client</c>.
     /// </summary>
     [System.Text.Json.Serialization.JsonPropertyName("priority_policy")]
     [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)]

src/NATS.Client.JetStream/Models/ConsumerGetnextRequest.cs

@@ -82,4 +82,17 @@ public record ConsumerGetnextRequest
     [System.Text.Json.Serialization.JsonPropertyName("id")]
     [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)]
     public string? Id { get; set; }
+
+    /// <summary>
+    /// Priority for message delivery when using prioritized priority policy.
+    /// </summary>
+    /// <remarks>
+    /// Lower values indicate higher priority (0 is the highest priority).
+    /// Maximum priority value is 9. This field is only used when the consumer
+    /// has PriorityPolicy set to "prioritized".
+    /// </remarks>
+    [System.Text.Json.Serialization.JsonPropertyName("priority")]
+    [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)]
+    [System.ComponentModel.DataAnnotations.Range(0, 9)]
+    public byte Priority { get; set; }
 }

src/NATS.Client.JetStream/NatsJSConsumer.cs

@@ -336,6 +336,7 @@ await sub.CallMsgNextAsync(
                 Group = opts.PriorityGroup?.Group,
                 MinPending = opts.PriorityGroup?.MinPending ?? 0,
                 MinAckPending = opts.PriorityGroup?.MinAckPending ?? 0,
+                Priority = opts.PriorityGroup?.Priority ?? 0,
             },
             cancellationToken).ConfigureAwait(false);
 
@@ -438,6 +439,7 @@ await sub.CallMsgNextAsync(
                     Group = opts.PriorityGroup?.Group,
                     MinPending = opts.PriorityGroup?.MinPending ?? 0,
                     MinAckPending = opts.PriorityGroup?.MinAckPending ?? 0,
+                    Priority = opts.PriorityGroup?.Priority ?? 0,
                 }
                 : new ConsumerGetnextRequest
                 {
@@ -449,6 +451,7 @@ await sub.CallMsgNextAsync(
                     Group = opts.PriorityGroup?.Group,
                     MinPending = opts.PriorityGroup?.MinPending ?? 0,
                     MinAckPending = opts.PriorityGroup?.MinAckPending ?? 0,
+                    Priority = opts.PriorityGroup?.Priority ?? 0,
                 },
             cancellationToken).ConfigureAwait(false);
 

src/NATS.Client.JetStream/NatsJSContext.Consumers.cs

@@ -269,9 +269,9 @@ private async ValueTask<NatsJSConsumer> CreateOrUpdateConsumerInternalAsync(
         }
 
         // TODO: enum these values?
-        if (config.PriorityPolicy != null && config.PriorityPolicy != "none" && config.PriorityPolicy != "overflow" && config.PriorityPolicy != "pinned_client")
+        if (config.PriorityPolicy != null && config.PriorityPolicy != "none" && config.PriorityPolicy != "overflow" && config.PriorityPolicy != "pinned_client" && config.PriorityPolicy != "prioritized")
         {
-            throw new NatsJSException("Cannot create consumers with priority policy other than 'overflow', 'pinned_client', or 'none'.");
+            throw new NatsJSException("Cannot create consumers with priority policy other than 'overflow', 'pinned_client', 'prioritized', or 'none'.");
         }
 
         var response = await JSRequestResponseAsync<ConsumerCreateRequest, ConsumerInfo>(

src/NATS.Client.JetStream/NatsJSOpts.cs

@@ -265,4 +265,14 @@ public record NatsJSPriorityGroupOpts
     /// When specified, this Pull request will only receive messages when the consumer has at least this many ack pending messages.
     /// </summary>
     public long MinAckPending { get; set; }
+
+    /// <summary>
+    /// Priority for message delivery when using prioritized priority policy.
+    /// </summary>
+    /// <remarks>
+    /// Lower values indicate higher priority (0 is the highest priority).
+    /// Maximum priority value is 9. This field is only used when the consumer
+    /// has PriorityPolicy set to "prioritized".
+    /// </remarks>
+    public byte Priority { get; init; }
 }

tests/NATS.Client.JetStream.Tests/PriorityGroupTest.cs

@@ -151,4 +151,87 @@ public async Task Consume_from_overflow_group()
             }
         }
     }
+
+    [SkipIfNatsServer(versionEarlierThan: "2.12")]
+    public async Task Fetch_from_prioritized_group_with_priority()
+    {
+        await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url });
+        var js = new NatsJSContext(nats);
+        var prefix = _server.GetNextId();
+
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+
+        await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.>"], cts.Token);
+
+        for (var i = 0; i < 10; i++)
+        {
+            var ack = await js.PublishAsync($"{prefix}s1.{i}", i, cancellationToken: cts.Token);
+            ack.EnsureSuccess();
+        }
+
+        var consumerConfig = new ConsumerConfig($"{prefix}c1")
+        {
+            PriorityGroups = ["jobs"],
+            PriorityPolicy = "prioritized",
+        };
+        var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", consumerConfig, cancellationToken: cts.Token);
+
+        // Test with priority 5
+        {
+            var opts = new NatsJSFetchOpts
+            {
+                MaxMsgs = 3,
+                PriorityGroup = new NatsJSPriorityGroupOpts { Group = "jobs", Priority = 5 },
+            };
+            var count = 0;
+            await foreach (var msg in consumer.FetchAsync<int>(opts, cancellationToken: cts.Token))
+            {
+                Assert.Equal(count++, msg.Data);
+                if (count == 3) break;
+            }
+
+            Assert.Equal(3, count);
+        }
+
+        // Test with priority 0 (highest priority)
+        {
+            var opts = new NatsJSFetchOpts
+            {
+                MaxMsgs = 2,
+                PriorityGroup = new NatsJSPriorityGroupOpts { Group = "jobs", Priority = 0 },
+            };
+            var count = 0;
+            await foreach (var msg in consumer.FetchAsync<int>(opts, cancellationToken: cts.Token))
+            {
+                Assert.Equal(count + 3, msg.Data); // Should continue from where we left off
+                count++;
+                if (count == 2) break;
+            }
+
+            Assert.Equal(2, count);
+        }
+    }
+
+    [SkipIfNatsServer(versionEarlierThan: "2.12")]
+    public async Task Consumer_with_prioritized_policy_validation()
+    {
+        await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url });
+        var js = new NatsJSContext(nats);
+        var prefix = _server.GetNextId();
+
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+
+        await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.>"], cts.Token);
+
+        // Test that prioritized policy is accepted
+        var consumerConfig = new ConsumerConfig($"{prefix}c1")
+        {
+            PriorityGroups = ["jobs"],
+            PriorityPolicy = "prioritized",
+        };
+
+        var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", consumerConfig, cancellationToken: cts.Token);
+        Assert.NotNull(consumer);
+        Assert.Equal("prioritized", consumer.Info.Config.PriorityPolicy);
+    }
 }