Add discard_new_per_subject to Stream

async-nats/src/jetstream/stream.rs

@@ -797,6 +797,9 @@ pub struct Config {
     /// When a Stream has reached its configured `max_bytes` or `max_msgs`, this policy kicks in.
     /// `DiscardPolicy::New` refuses new messages or `DiscardPolicy::Old` (default) deletes old messages to make space
     pub discard: DiscardPolicy,
+    /// Prevents a message from being added to a stream if the max_msgs_per_subject limit for the subject has been reached
+    #[serde(default, skip_serializing_if = "is_default")]
+    pub discard_new_per_subject: bool,
     /// Which NATS subjects to populate this stream with. Supports wildcards. Defaults to just the
     /// configured stream `name`.
     #[serde(default, skip_serializing_if = "Vec::is_empty")]

async-nats/tests/jetstream_tests.rs

@@ -1843,4 +1843,38 @@ mod jetstream {
             message.ack().await.unwrap();
         }
     }
+
+    #[tokio::test]
+    async fn discard_new_per_subject() {
+        let server = nats_server::run_server("tests/configs/jetstream.conf");
+        let client = async_nats::connect(server.client_url()).await.unwrap();
+
+        let jetstream = async_nats::jetstream::new(client);
+
+        let _source_stream = jetstream
+            .create_stream(async_nats::jetstream::stream::Config {
+                name: "source".to_string(),
+                max_messages: 10,
+                max_messages_per_subject: 2,
+                discard_new_per_subject: true,
+                subjects: vec!["events.>".to_string()],
+                discard: DiscardPolicy::New,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        jetstream
+            .publish("events.1".to_string(), "data".into())
+            .await
+            .unwrap();
+        jetstream
+            .publish("events.1".to_string(), "data".into())
+            .await
+            .unwrap();
+        jetstream
+            .publish("events.1".to_string(), "data".into())
+            .await
+            .expect_err("should get 503 maximum messages per subject exceeded error");
+    }
 }