Skip to content

Commit

Permalink
Add discard_new_per_subject to Stream
Browse files Browse the repository at this point in the history
  • Loading branch information
Jarema committed Oct 3, 2022
1 parent 2e9425e commit 93b4072
Show file tree
Hide file tree
Showing 2 changed files with 37 additions and 0 deletions.
3 changes: 3 additions & 0 deletions async-nats/src/jetstream/stream.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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")]
Expand Down
34 changes: 34 additions & 0 deletions async-nats/tests/jetstream_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1846,4 +1846,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");
}
}

0 comments on commit 93b4072

Please sign in to comment.