New command group: streams

Author: michaelklishin

CHANGELOG.md

@@ -8,6 +8,7 @@
  * `nodes` is a new command group for operations on nodes
  * `parameters` is a new command group for operations on [runtime parameters](https://www.rabbitmq.com/docs/parameters)
  * `queues` is a new command group for operations on queues
+ * `streams` is a new command group for operations on streams
  * `users` is a new command group for operations on users
  * `vhosts` is a new command group for operations on virtual hosts
  * Command groups are now ordered alphabetically

src/cli.rs

@@ -236,6 +236,12 @@ pub fn parser(pre_flight_settings: PreFlightSettings) -> Command {
         ))
         .subcommand_value_name("shovels")
         .subcommands(shovel_subcommands(pre_flight_settings.clone()));
+    let streams_group = Command::new("streams")
+        .about("Operations on streams")
+        .infer_subcommands(pre_flight_settings.infer_subcommands)
+        .infer_long_args(pre_flight_settings.infer_long_options)
+        .subcommand_value_name("stream")
+        .subcommands(streams_subcommands(pre_flight_settings.clone()));
     let tanzu_group = Command::new("tanzu")
         .about("Tanzu RabbitMQ-specific commands")
         .infer_subcommands(pre_flight_settings.infer_subcommands)
@@ -285,6 +291,7 @@ pub fn parser(pre_flight_settings: PreFlightSettings) -> Command {
         rebalance_group,
         show_group,
         shovels_group,
+        streams_group,
         tanzu_group,
         users_group,
         vhosts_group,
@@ -1306,6 +1313,68 @@ fn queues_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 5] {
         .map(|cmd| cmd.infer_long_args(pre_flight_settings.infer_long_options))
 }
 
+fn streams_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 3] {
+    let declare_cmd = Command::new("declare")
+        .about("Declares a stream")
+        .after_help(color_print::cformat!(
+            "<bold>Doc guide:</bold>: {}",
+            STREAM_GUIDE_URL
+        ))
+        .arg(Arg::new("name").long("name").required(true).help("name"))
+        .arg(
+            Arg::new("expiration")
+                .long("expiration")
+                .help("stream expiration, e.g. 12h for 12 hours, 7D for 7 days, or 1M for 1 month")
+                .required(true)
+                .value_parser(value_parser!(String)),
+        )
+        .arg(
+            Arg::new("max_length_bytes")
+                .long("max-length-bytes")
+                .help("maximum stream length in bytes")
+                .required(false)
+                .value_parser(value_parser!(u64)),
+        )
+        .arg(
+            Arg::new("max_segment_length_bytes")
+                .long("stream-max-segment-size-bytes")
+                .help("maximum stream segment file length in bytes")
+                .required(false)
+                .value_parser(value_parser!(u64)),
+        )
+        .arg(
+            Arg::new("arguments")
+                .long("arguments")
+                .help("additional exchange arguments")
+                .required(false)
+                .default_value("{}")
+                .value_parser(value_parser!(String)),
+        );
+    let idempotently_arg = Arg::new("idempotently")
+        .long("idempotently")
+        .value_parser(value_parser!(bool))
+        .action(ArgAction::SetTrue)
+        .help("do not consider 404 Not Found API responses to be errors")
+        .required(false);
+    let delete_cmd = Command::new("delete")
+        .about("Deletes a queue")
+        .arg(
+            Arg::new("name")
+                .long("name")
+                .help("queue name")
+                .required(true),
+        )
+        .arg(idempotently_arg.clone());
+    let list_cmd = Command::new("list")
+        .long_about("Lists streams and queues and")
+        .after_help(color_print::cformat!(
+            "<bold>Doc guide</bold>: {}",
+            STREAM_GUIDE_URL
+        ));
+    [declare_cmd, delete_cmd, list_cmd]
+        .map(|cmd| cmd.infer_long_args(pre_flight_settings.infer_long_options))
+}
+
 fn parameters_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 3] {
     let list_cmd = Command::new("list")
         .arg(

src/main.rs

@@ -730,6 +730,18 @@ fn dispatch_common_subcommand(
             let result = commands::list_shovels(client);
             res_handler.tabular_result(result)
         }
+        ("streams", "declare") => {
+            let result = commands::declare_stream(client, &vhost, second_level_args);
+            res_handler.no_output_on_success(result);
+        }
+        ("streams", "delete") => {
+            let result = commands::delete_queue(client, &vhost, second_level_args);
+            res_handler.delete_operation_result(result);
+        }
+        ("streams", "list") => {
+            let result = commands::list_queues(client, &vhost);
+            res_handler.tabular_result(result)
+        }
         ("users", "connections") => {
             let result = commands::list_user_connections(client, second_level_args);
             res_handler.tabular_result(result)

tests/streams_tests.rs

@@ -74,3 +74,61 @@ fn list_streams() -> Result<(), Box<dyn std::error::Error>> {
 
     Ok(())
 }
+
+#[test]
+fn streams_list() -> Result<(), Box<dyn std::error::Error>> {
+    let vh1 = "stream_vhost_3";
+    let vh2 = "stream_vhost_4";
+    let s1 = "new_stream1";
+    let s2 = "new_stream2";
+
+    delete_vhost(vh1).expect("failed to delete a virtual host");
+    delete_vhost(vh2).expect("failed to delete a virtual host");
+
+    // declare vhost 1
+    run_succeeds(["vhosts", "declare", "--name", vh1]);
+
+    // declare vhost 2
+    run_succeeds(["vhosts", "declare", "--name", vh2]);
+
+    // declare a new stream in vhost 1
+    run_succeeds([
+        "-V",
+        vh1,
+        "streams",
+        "declare",
+        "--name",
+        s1,
+        "--expiration",
+        "2D",
+    ]);
+
+    // declare new stream in vhost 2
+    run_succeeds([
+        "-V",
+        vh2,
+        "streams",
+        "declare",
+        "--name",
+        s2,
+        "--expiration",
+        "12h",
+    ]);
+
+    await_queue_metric_emission();
+
+    // list streams in vhost 1
+    run_succeeds(["-V", vh1, "streams", "list"])
+        .stdout(predicate::str::contains(s1).and(predicate::str::contains("random_stream").not()));
+
+    // delete the stream in vhost 1
+    run_succeeds(["-V", vh1, "streams", "delete", "--name", s1]);
+
+    // list streams in vhost 1
+    run_succeeds(["-V", vh1, "streams", "list"]).stdout(predicate::str::contains(s1).not());
+
+    delete_vhost(vh1).expect("failed to delete a virtual host");
+    delete_vhost(vh2).expect("failed to delete a virtual host");
+
+    Ok(())
+}