Allow optional numeric ID for string and topic create command (#571)

iggy-rs · Jan 19, 2024 · e8ea17f · e8ea17f
1 parent fe02d1d
commit e8ea17f
Show file tree

Hide file tree

Showing 43 changed files with 259 additions and 120 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/bench/src/benchmarks/benchmark.rs b/bench/src/benchmarks/benchmark.rs
@@ -70,7 +70,10 @@ pub trait Benchmarkable {
                 info!("Creating the test stream {}", stream_id);
                 let name = format!("stream {}", stream_id);
                 client
-                    .create_stream(&CreateStream { stream_id, name })
+                    .create_stream(&CreateStream {
+                        stream_id: Some(stream_id),
+                        name,
+                    })
                     .await?;
 
                 info!(
@@ -81,7 +84,7 @@ pub trait Benchmarkable {
                 client
                     .create_topic(&CreateTopic {
                         stream_id: Identifier::numeric(stream_id)?,
-                        topic_id,
+                        topic_id: Some(topic_id),
                         partitions_count,
                         name,
                         message_expiry: None,

diff --git a/examples/Cargo.toml b/examples/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "iggy_examples"
-version = "0.0.1"
+version = "0.0.2"
 edition = "2021"
 
 [[example]]

diff --git a/examples/src/getting-started/producer/main.rs b/examples/src/getting-started/producer/main.rs
@@ -50,7 +50,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 async fn init_system(client: &IggyClient) {
     match client
         .create_stream(&CreateStream {
-            stream_id: STREAM_ID,
+            stream_id: Some(STREAM_ID),
             name: "sample-stream".to_string(),
         })
         .await
@@ -62,7 +62,7 @@ async fn init_system(client: &IggyClient) {
     match client
         .create_topic(&CreateTopic {
             stream_id: Identifier::numeric(STREAM_ID).unwrap(),
-            topic_id: TOPIC_ID,
+            topic_id: Some(TOPIC_ID),
             partitions_count: 1,
             name: "sample-topic".to_string(),
             message_expiry: None,

diff --git a/examples/src/shared/system.rs b/examples/src/shared/system.rs
@@ -79,14 +79,14 @@ pub async fn init_by_producer(args: &Args, client: &dyn Client) -> Result<(), Er
     info!("Stream does not exist, creating...");
     client
         .create_stream(&CreateStream {
-            stream_id: args.stream_id,
+            stream_id: Some(args.stream_id),
             name: "sample".to_string(),
         })
         .await?;
     client
         .create_topic(&CreateTopic {
             stream_id: Identifier::numeric(args.stream_id).unwrap(),
-            topic_id: args.topic_id,
+            topic_id: Some(args.topic_id),
             partitions_count: args.partitions_count,
             name: "orders".to_string(),
             message_expiry: None,

diff --git a/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs b/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
@@ -71,7 +71,7 @@ impl IggyCmdTestCase for TestConsumerGroupCreateCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -80,7 +80,7 @@ impl IggyCmdTestCase for TestConsumerGroupCreateCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: 0,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/consumer_group/test_consumer_group_delete_command.rs b/integration/tests/cli/consumer_group/test_consumer_group_delete_command.rs
@@ -77,7 +77,7 @@ impl IggyCmdTestCase for TestConsumerGroupDeleteCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -86,7 +86,7 @@ impl IggyCmdTestCase for TestConsumerGroupDeleteCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: 0,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs b/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs
@@ -77,7 +77,7 @@ impl IggyCmdTestCase for TestConsumerGroupGetCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -86,7 +86,7 @@ impl IggyCmdTestCase for TestConsumerGroupGetCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: 0,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/consumer_group/test_consumer_group_list_command.rs b/integration/tests/cli/consumer_group/test_consumer_group_list_command.rs
@@ -74,7 +74,7 @@ impl IggyCmdTestCase for TestConsumerGroupListCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -83,7 +83,7 @@ impl IggyCmdTestCase for TestConsumerGroupListCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: 0,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/message/test_message_poll_command.rs b/integration/tests/cli/message/test_message_poll_command.rs
@@ -99,7 +99,7 @@ impl IggyCmdTestCase for TestMessagePollCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -108,7 +108,7 @@ impl IggyCmdTestCase for TestMessagePollCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: self.partitions_count,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/message/test_message_send_command.rs b/integration/tests/cli/message/test_message_send_command.rs
@@ -128,7 +128,7 @@ impl IggyCmdTestCase for TestMessageSendCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -137,7 +137,7 @@ impl IggyCmdTestCase for TestMessageSendCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: self.partitions_count,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/partition/test_partition_create_command.rs b/integration/tests/cli/partition/test_partition_create_command.rs
@@ -70,7 +70,7 @@ impl IggyCmdTestCase for TestPartitionCreateCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -79,7 +79,7 @@ impl IggyCmdTestCase for TestPartitionCreateCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: self.partitions_count,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/partition/test_partition_delete_command.rs b/integration/tests/cli/partition/test_partition_delete_command.rs
@@ -70,7 +70,7 @@ impl IggyCmdTestCase for TestPartitionDeleteCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -79,7 +79,7 @@ impl IggyCmdTestCase for TestPartitionDeleteCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: self.partitions_count,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/stream/test_stream_delete_command.rs b/integration/tests/cli/stream/test_stream_delete_command.rs
@@ -38,7 +38,7 @@ impl IggyCmdTestCase for TestStreamDeleteCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.name.clone(),
             })
             .await;

diff --git a/integration/tests/cli/stream/test_stream_get_command.rs b/integration/tests/cli/stream/test_stream_get_command.rs
@@ -37,7 +37,7 @@ impl IggyCmdTestCase for TestStreamGetCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.name.clone(),
             })
             .await;

diff --git a/integration/tests/cli/stream/test_stream_list_command.rs b/integration/tests/cli/stream/test_stream_list_command.rs
@@ -34,7 +34,7 @@ impl IggyCmdTestCase for TestStreamListCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.name.clone(),
             })
             .await;

diff --git a/integration/tests/cli/stream/test_stream_update_command.rs b/integration/tests/cli/stream/test_stream_update_command.rs
@@ -42,7 +42,7 @@ impl IggyCmdTestCase for TestStreamUpdateCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.name.clone(),
             })
             .await;

diff --git a/integration/tests/cli/system/test_stats_command.rs b/integration/tests/cli/system/test_stats_command.rs
@@ -15,15 +15,15 @@ impl IggyCmdTestCase for TestStatsCmd {
         let stream_id = Identifier::from_str_value("logs").unwrap();
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: 1,
+                stream_id: Some(1),
                 name: stream_id.as_string(),
             })
             .await;
         assert!(stream.is_ok());
 
         let topic = client
             .create_topic(&CreateTopic {
-                topic_id: 1,
+                topic_id: Some(1),
                 stream_id,
                 partitions_count: 5,
                 message_expiry: None,

diff --git a/integration/tests/cli/topic/test_topic_create_command.rs b/integration/tests/cli/topic/test_topic_create_command.rs
@@ -72,7 +72,7 @@ impl IggyCmdTestCase for TestTopicCreateCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;

diff --git a/integration/tests/cli/topic/test_topic_delete_command.rs b/integration/tests/cli/topic/test_topic_delete_command.rs
@@ -59,7 +59,7 @@ impl IggyCmdTestCase for TestTopicDeleteCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -68,7 +68,7 @@ impl IggyCmdTestCase for TestTopicDeleteCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: 1,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/topic/test_topic_get_command.rs b/integration/tests/cli/topic/test_topic_get_command.rs
@@ -60,7 +60,7 @@ impl IggyCmdTestCase for TestTopicGetCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -69,7 +69,7 @@ impl IggyCmdTestCase for TestTopicGetCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: 1,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/topic/test_topic_list_command.rs b/integration/tests/cli/topic/test_topic_list_command.rs
@@ -57,7 +57,7 @@ impl IggyCmdTestCase for TestTopicListCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -66,7 +66,7 @@ impl IggyCmdTestCase for TestTopicListCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: 1,
                 name: self.topic_name.clone(),
                 message_expiry: None,

diff --git a/integration/tests/cli/topic/test_topic_update_command.rs b/integration/tests/cli/topic/test_topic_update_command.rs
@@ -104,7 +104,7 @@ impl IggyCmdTestCase for TestTopicUpdateCmd {
     async fn prepare_server_state(&mut self, client: &dyn Client) {
         let stream = client
             .create_stream(&CreateStream {
-                stream_id: self.stream_id,
+                stream_id: Some(self.stream_id),
                 name: self.stream_name.clone(),
             })
             .await;
@@ -125,7 +125,7 @@ impl IggyCmdTestCase for TestTopicUpdateCmd {
         let topic = client
             .create_topic(&CreateTopic {
                 stream_id: Identifier::numeric(self.stream_id).unwrap(),
-                topic_id: self.topic_id,
+                topic_id: Some(self.topic_id),
                 partitions_count: 1,
                 name: self.topic_name.clone(),
                 message_expiry,

diff --git a/integration/tests/examples/mod.rs b/integration/tests/examples/mod.rs
@@ -122,15 +122,15 @@ impl<'a> IggyExampleTest<'a> {
         if existing_stream_and_topic {
             self.client
                 .create_stream(&CreateStream {
-                    stream_id: 1,
+                    stream_id: Some(1),
                     name: "sample-stream".to_string(),
                 })
                 .await
                 .unwrap();
             self.client
                 .create_topic(&CreateTopic {
                     stream_id: Identifier::numeric(1).unwrap(),
-                    topic_id: 1,
+                    topic_id: Some(1),
                     partitions_count: 1,
                     name: "sample-topic".to_string(),
                     message_expiry: None,

diff --git a/integration/tests/examples/test_getting_started.rs b/integration/tests/examples/test_getting_started.rs
@@ -57,9 +57,9 @@ async fn should_succeed_with_preexisting_stream_and_topic() {
     iggy_example_test
         .execute_test(TestGettingStarted {
             expected_producer_output: vec![
-                "Received an invalid response with status: 1011 (stream_id_already_exists).",
+                "Received an invalid response with status: 1012 (stream_name_already_exists).",
                 "Stream already exists and will not be created again.",
-                "Received an invalid response with status: 2012 (topic_id_already_exists).",
+                "Received an invalid response with status: 2013 (topic_name_already_exists).",
                 "Topic already exists and will not be created again.",
                 "Sent 10 message(s).",
             ],