New command group: queues

Author: michaelklishin

CHANGELOG.md

@@ -7,6 +7,7 @@
  * `global_parameters` is a new command group for operations on [global runtime parameters](https://www.rabbitmq.com/docs/parameters)
  * `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
  * `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

@@ -201,6 +201,12 @@ pub fn parser(pre_flight_settings: PreFlightSettings) -> Command {
         .infer_long_args(pre_flight_settings.infer_long_options)
         .subcommand_value_name("queue")
         .subcommands(purge_subcommands(pre_flight_settings.clone()));
+    let queues_group = Command::new("queues")
+        .about("Operations on queues")
+        .infer_subcommands(pre_flight_settings.infer_subcommands)
+        .infer_long_args(pre_flight_settings.infer_long_options)
+        .subcommand_value_name("queue")
+        .subcommands(queues_subcommands(pre_flight_settings.clone()));
     let rebalance_group = Command::new("rebalance")
         .about("Rebalancing of leader replicas")
         .infer_subcommands(pre_flight_settings.infer_subcommands)
@@ -275,6 +281,7 @@ pub fn parser(pre_flight_settings: PreFlightSettings) -> Command {
         policies_group,
         publish_group,
         purge_group,
+        queues_group,
         rebalance_group,
         show_group,
         shovels_group,
@@ -1227,6 +1234,78 @@ fn purge_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 1] {
     [queue_cmd].map(|cmd| cmd.infer_long_args(pre_flight_settings.infer_long_options))
 }
 
+fn queues_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 5] {
+    let declare_cmd = Command::new("declare")
+        .about("Declares a queue or a stream")
+        .after_help(color_print::cformat!(
+            "<bold>Doc guide:</bold>: {}",
+            QUEUE_GUIDE_URL
+        ))
+        .arg(Arg::new("name").long("name").required(true).help("name"))
+        .arg(
+            Arg::new("type")
+                .long("type")
+                .help("queue type")
+                .value_parser(value_parser!(QueueType))
+                .required(false)
+                .default_value("classic"),
+        )
+        .arg(
+            Arg::new("durable")
+                .long("durable")
+                .help("should it persist after a restart")
+                .required(false)
+                .value_parser(value_parser!(bool)),
+        )
+        .arg(
+            Arg::new("auto_delete")
+                .long("auto-delete")
+                .help("should it be deleted when the last consumer disconnects")
+                .required(false)
+                .value_parser(value_parser!(bool)),
+        )
+        .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 queues and streams")
+        .after_help(color_print::cformat!(
+            "<bold>Doc guide</bold>: {}",
+            QUEUE_GUIDE_URL
+        ));
+    let purge_cmd = Command::new("purge")
+        .long_about("Purges (permanently removes unacknowledged messages from) a queue")
+        .arg(
+            Arg::new("name")
+                .long("name")
+                .help("name of the queue to purge")
+                .required(true),
+        );
+    let rebalance_cmd = Command::new("rebalance").about("Rebalances queue leaders");
+    [declare_cmd, delete_cmd, list_cmd, purge_cmd, rebalance_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

@@ -641,6 +641,26 @@ fn dispatch_common_subcommand(
             let result = commands::purge_queue(client, &vhost, second_level_args);
             res_handler.no_output_on_success(result);
         }
+        ("queues", "declare") => {
+            let result = commands::declare_queue(client, &vhost, second_level_args);
+            res_handler.no_output_on_success(result);
+        }
+        ("queues", "delete") => {
+            let result = commands::delete_queue(client, &vhost, second_level_args);
+            res_handler.delete_operation_result(result);
+        }
+        ("queues", "list") => {
+            let result = commands::list_queues(client, &vhost);
+            res_handler.tabular_result(result)
+        }
+        ("queues", "purge") => {
+            let result = commands::purge_queue(client, &vhost, second_level_args);
+            res_handler.no_output_on_success(result);
+        }
+        ("queues", "rebalance") => {
+            let result = commands::rebalance_queues(client);
+            res_handler.no_output_on_success(result);
+        }
         ("rebalance", "queues") => {
             let result = commands::rebalance_queues(client);
             res_handler.no_output_on_success(result);

tests/queues_tests.rs

@@ -49,7 +49,10 @@ fn list_queues() -> Result<(), Box<dyn std::error::Error>> {
     run_succeeds(["-V", vh1, "list", "queues"])
         .stdout(predicate::str::contains(q1).and(predicate::str::contains("new_queue2").not()));
 
-    // delete the queue in vhost 1
+    // purge a queue in vhost 1
+    run_succeeds(["-V", vh1, "purge", "queue", "--name", q1]);
+
+    // delete a queue in vhost 1
     run_succeeds(["-V", vh1, "delete", "queue", "--name", q1]);
 
     // list queues in vhost 1
@@ -60,3 +63,50 @@ fn list_queues() -> Result<(), Box<dyn std::error::Error>> {
 
     Ok(())
 }
+
+#[test]
+fn queues_lists() -> Result<(), Box<dyn std::error::Error>> {
+    let vh1 = "queue_vhost_3";
+    let vh2 = "queue_vhost_4";
+    let q1 = "new_queue1";
+    let q2 = "new_queue2";
+
+    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 queue in vhost 1
+    run_succeeds([
+        "-V", vh1, "queues", "declare", "--name", q1, "--type", "classic",
+    ]);
+
+    // declare new queue in vhost 2
+    run_succeeds([
+        "-V", vh2, "queues", "declare", "--name", q2, "--type", "quorum",
+    ]);
+
+    await_queue_metric_emission();
+
+    // list queues in vhost 1
+    run_succeeds(["-V", vh1, "queues", "list"])
+        .stdout(predicate::str::contains(q1).and(predicate::str::contains("new_queue2").not()));
+
+    // purge a queue in vhost 1
+    run_succeeds(["-V", vh1, "queues", "purge", "--name", q1]);
+
+    // delete a queue in vhost 1
+    run_succeeds(["-V", vh1, "queues", "delete", "--name", q1]);
+
+    // list queues in vhost 1
+    run_succeeds(["-V", vh1, "queues", "list"]).stdout(predicate::str::contains(q1).not());
+
+    delete_vhost(vh1).expect("failed to delete a virtual host");
+    delete_vhost(vh2).expect("failed to delete a virtual host");
+
+    Ok(())
+}