Change ConsumerConfig AckPolicy default to Explicit (#490)

* Change ConsumerConfig constructor to always set default AckPolicy

* Update default AckPolicy for ConsumerConfig

The default constructor for the ConsumerConfig class no longer sets
the AckPolicy property to Explicit. Also, the values of the
ConsumerConfigAckPolicy enumeration have been rearranged, making
Explicit the default value (0).

* Removed redundant default ACK policy assignments

* Fixed consumer ack policy default

* fixed test

when ack policy is set to explicit, the default
ack wait seems to be set to 30s by the server.

---------

Co-authored-by: Ziya Suzen <ziya@suzen.net>

Author: hanlong-chen-1047

sandbox/Example.JetStream.PullConsumer/Program.cs

@@ -19,7 +19,7 @@
 
 var js = new NatsJSContext(nats);
 
-var consumer = await js.CreateOrUpdateConsumerAsync("s1", new ConsumerConfig { Name = "c1", DurableName = "c1", AckPolicy = ConsumerConfigAckPolicy.Explicit });
+var consumer = await js.CreateOrUpdateConsumerAsync("s1", new ConsumerConfig { Name = "c1", DurableName = "c1" });
 
 var idle = TimeSpan.FromSeconds(5);
 var expires = TimeSpan.FromSeconds(10);

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

@@ -29,7 +29,6 @@ public ConsumerConfig(string name)
     {
         Name = name;
         DurableName = name;
-        AckPolicy = ConsumerConfigAckPolicy.Explicit;
     }
 
     [System.Text.Json.Serialization.JsonPropertyName("deliver_policy")]
@@ -91,7 +90,7 @@ public ConsumerConfig(string name)
 #if NET6_0
     [System.Text.Json.Serialization.JsonConverter(typeof(NatsJSJsonStringEnumConverter<ConsumerConfigAckPolicy>))]
 #endif
-    public ConsumerConfigAckPolicy AckPolicy { get; set; } = ConsumerConfigAckPolicy.None;
+    public ConsumerConfigAckPolicy AckPolicy { get; set; } = ConsumerConfigAckPolicy.Explicit;
 
     /// <summary>
     /// How long (in nanoseconds) to allow messages to remain un-acknowledged before attempting redelivery

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

@@ -2,7 +2,7 @@ namespace NATS.Client.JetStream.Models;
 
 public enum ConsumerConfigAckPolicy
 {
-    None = 0,
+    Explicit = 0,
     All = 1,
-    Explicit = 2,
+    None = 2,
 }

tests/NATS.Client.CheckNativeAot/Program.cs

@@ -86,9 +86,6 @@ async Task JetStreamTests()
             {
                 Name = "consumer1",
                 DurableName = "consumer1",
-
-                // Turn on ACK so we can test them below
-                AckPolicy = ConsumerConfigAckPolicy.Explicit,
             },
             cts1.Token);
         AssertEqual("events", consumer.Info.StreamName);

tests/NATS.Client.Core.MemoryTests/NatsConsumeTests.cs

@@ -29,7 +29,7 @@ public void Subscription_should_not_be_collected_when_in_consume_async_enumerato
             {
                 await js.CreateStreamAsync(new StreamConfig { Name = "s1", Subjects = new[] { "s1.*" } });
 
-                var consumer = await js.CreateOrUpdateConsumerAsync("s1", new ConsumerConfig { Name = "c1", DurableName = "c1", AckPolicy = ConsumerConfigAckPolicy.Explicit });
+                var consumer = await js.CreateOrUpdateConsumerAsync("s1", new ConsumerConfig { Name = "c1", DurableName = "c1" });
 
                 var count = 0;
                 await foreach (var msg in consumer.ConsumeAsync<int>(opts: new NatsJSConsumeOpts { MaxMsgs = 100 }))

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

@@ -76,9 +76,6 @@ public async Task Create_stream_test()
                 {
                     Name = "consumer1",
                     DurableName = "consumer1",
-
-                    // Turn on ACK so we can test them below
-                    AckPolicy = ConsumerConfigAckPolicy.Explicit,
                 },
                 cts1.Token);
             Assert.Equal("events", consumer.Info.StreamName);

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

@@ -148,7 +148,7 @@ public async Task Consumer_create_update_action()
         // Update consumer
         {
             var c1 = await js.GetConsumerAsync("s1", "c1");
-            Assert.Equal(default, c1.Info.Config.AckWait);
+            Assert.Equal(TimeSpan.FromSeconds(30), c1.Info.Config.AckWait);
 
             var changedConsumerConfig = new ConsumerConfig { Name = "c1", AckWait = TimeSpan.FromSeconds(10) };
             await js.UpdateConsumerAsync("s1", changedConsumerConfig);

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

@@ -25,4 +25,16 @@ public void Placement_properties_should_be_optional()
         Assert.Null(result.Cluster);
         Assert.Null(result.Tags);
     }
+
+    [Fact]
+    public void Default_consumer_ack_policy_should_be_explicit()
+    {
+        var serializer = NatsJSJsonSerializer<ConsumerConfig>.Default;
+
+        var bw = new NatsBufferWriter<byte>();
+        serializer.Serialize(bw, new ConsumerConfig());
+
+        var json = Encoding.UTF8.GetString(bw.WrittenSpan);
+        Assert.Matches("\"ack_policy\":\"explicit\"", json);
+    }
 }

tests/NATS.Net.DocsExamples/JetStream/ManagingPage.cs

@@ -70,7 +70,6 @@ public async Task Run()
             {
                 Name = "durable_processor",
                 DurableName = "durable_processor",
-                AckPolicy = ConsumerConfigAckPolicy.Explicit,
             };
 
             var consumer = await js.CreateOrUpdateConsumerAsync(stream: "orders", durableConfig);