This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 7da163362 fix(sdk): removes topic_id and stream_id from public Stream
API (#4249)
7da163362 is described below
commit 7da163362572d60feadd6b012baec37b96b95ea8
Author: haubur <[email protected]>
AuthorDate: Sat Oct 10 17:54:17 2026 +0200
fix(sdk): removes topic_id and stream_id from public Stream API (#4249)
---
core/connectors/runtime/src/source.rs | 4 +-
core/integration/tests/sdk/consumer_offset.rs | 2 -
core/integration/tests/sdk/mod.rs | 1 +
core/integration/tests/sdk/stream_builder.rs | 70 ++++++++++++++++++++++
core/sdk/src/clients/client.rs | 15 ++---
core/sdk/src/clients/consumer.rs | 44 +++++++-------
core/sdk/src/clients/producer.rs | 57 ++++++++++--------
.../src/stream_builder/build/build_stream_topic.rs | 42 ++++---------
.../stream_builder/config/config_iggy_consumer.rs | 58 +-----------------
.../stream_builder/config/config_iggy_producer.rs | 57 +-----------------
.../stream_builder/config/config_iggy_stream.rs | 32 +++-------
examples/rust/src/shared/system.rs | 10 ++--
.../rust/src/stream-builder/stream-basic/main.rs | 3 +-
.../stream-builder/stream-consumer-config/main.rs | 6 +-
.../stream-builder/stream-producer-config/main.rs | 6 +-
15 files changed, 172 insertions(+), 235 deletions(-)
diff --git a/core/connectors/runtime/src/source.rs
b/core/connectors/runtime/src/source.rs
index a6a747b2c..5722f2faa 100644
--- a/core/connectors/runtime/src/source.rs
+++ b/core/connectors/runtime/src/source.rs
@@ -505,12 +505,12 @@ async fn ensure_durable_source_topic(
stream_name: &str,
topic_name: &str,
) -> Result<(), RuntimeError> {
- let stream_id = Identifier::try_from(stream_name)?;
+ let stream_id = Identifier::named(stream_name)?;
if client.get_stream(&stream_id).await?.is_none() {
client.create_stream(stream_name).await?;
}
- let topic_id = Identifier::try_from(topic_name)?;
+ let topic_id = Identifier::named(topic_name)?;
let topic = match client.get_topic(&stream_id, &topic_id).await? {
Some(topic) => topic,
None => {
diff --git a/core/integration/tests/sdk/consumer_offset.rs
b/core/integration/tests/sdk/consumer_offset.rs
index c49cd16af..b12d2b4a3 100644
--- a/core/integration/tests/sdk/consumer_offset.rs
+++ b/core/integration/tests/sdk/consumer_offset.rs
@@ -131,9 +131,7 @@ async fn
given_standalone_stream_consumer_when_creating_topic_should_select_part
let topic_name = format!("{TOPIC_NAME}-{partitions_count}");
let topic_id = Identifier::named(&topic_name).unwrap();
let config = IggyConsumerConfig::builder()
- .stream_id(stream_id.clone())
.stream_name(STREAM_NAME)
- .topic_id(topic_id.clone())
.topic_name(topic_name)
.consumer_name(CONSUMER_NAME)
.consumer_kind(ConsumerKind::Consumer)
diff --git a/core/integration/tests/sdk/mod.rs
b/core/integration/tests/sdk/mod.rs
index b9b329866..483aff0c5 100644
--- a/core/integration/tests/sdk/mod.rs
+++ b/core/integration/tests/sdk/mod.rs
@@ -28,3 +28,4 @@ mod producer;
mod protocol_version;
mod raw;
mod send_confirmation;
+mod stream_builder;
diff --git a/core/integration/tests/sdk/stream_builder.rs
b/core/integration/tests/sdk/stream_builder.rs
new file mode 100644
index 000000000..2632dea42
--- /dev/null
+++ b/core/integration/tests/sdk/stream_builder.rs
@@ -0,0 +1,70 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use iggy::prelude::*;
+use iggy::stream_builder::{IggyConsumerConfig, IggyStreamConsumer};
+use integration::iggy_harness;
+
+const STREAM_NAME: &str = "123";
+const TOPIC_NAME: &str = "stream-builder-topic";
+const CONSUMER_NAME: &str = "stream-builder-consumer";
+
+#[iggy_harness]
+async fn
given_consumer_config_when_building_twice_should_reuse_the_named_stream_and_topic(
+ harness: &TestHarness,
+) {
+ let client = harness.root_client().await.expect("Root client");
+ let config = consumer_config();
+ let stream_id = Identifier::named(STREAM_NAME).unwrap();
+
+ for build in 1..=2 {
+ let mut consumer = IggyStreamConsumer::build(&client, &config)
+ .await
+ .unwrap_or_else(|err| {
+ panic!("Stream consumer should initialize on build {build}:
{err}")
+ });
+
+ let streams = client.get_streams().await.expect("Read streams");
+ let stream_names: Vec<&str> = streams.iter().map(|stream|
stream.name.as_str()).collect();
+ assert_eq!(stream_names, vec![STREAM_NAME]);
+
+ let topics = client.get_topics(&stream_id).await.expect("Read topics");
+ let topic_names: Vec<&str> = topics.iter().map(|topic|
topic.name.as_str()).collect();
+ assert_eq!(topic_names, vec![TOPIC_NAME]);
+
+ consumer.shutdown().await.expect("Shut down consumer");
+ }
+}
+
+fn consumer_config() -> IggyConsumerConfig {
+ IggyConsumerConfig::builder()
+ .stream_name(STREAM_NAME)
+ .topic_name(TOPIC_NAME)
+ .consumer_name(CONSUMER_NAME)
+ .consumer_kind(ConsumerKind::Consumer)
+ .auto_commit(AutoCommit::Disabled)
+ .batch_length(1)
+ .create_stream_if_not_exists(true)
+ .create_topic_if_not_exists(true)
+ .partitions_count(1)
+ .polling_interval("1ms".parse().unwrap())
+ .polling_strategy(PollingStrategy::first())
+ .polling_retry_interval(NonZeroIggyDuration::ONE_SECOND)
+ .init_retries(0)
+ .init_interval(NonZeroIggyDuration::ONE_SECOND)
+ .build()
+}
diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index 99c4dc2b8..8f7adc5e8 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -21,6 +21,7 @@ use crate::clients::client_builder::IggyClientBuilder;
use crate::http::http_client::HttpClient;
use crate::http::http_transport::HttpTransport;
use crate::prelude::EncryptorKind;
+use crate::prelude::Identifier;
use crate::prelude::IggyConsumerBuilder;
use crate::prelude::IggyError;
use crate::prelude::IggyProducerBuilder;
@@ -653,8 +654,8 @@ impl IggyClient {
self.client.clone(),
name.to_owned(),
Consumer::new(name.try_into()?),
- stream.try_into()?,
- topic.try_into()?,
+ Identifier::named(stream)?,
+ Identifier::named(topic)?,
Some(partition),
self.encryptor.clone(),
None,
@@ -723,9 +724,9 @@ impl IggyClient {
Ok(IggyConsumerBuilder::new(
self.client.clone(),
name.to_owned(),
- Consumer::group(name.try_into()?),
- stream.try_into()?,
- topic.try_into()?,
+ Consumer::group(Identifier::named(name)?),
+ Identifier::named(stream)?,
+ Identifier::named(topic)?,
None,
self.encryptor.clone(),
None,
@@ -784,9 +785,9 @@ impl IggyClient {
pub fn producer(&self, stream: &str, topic: &str) ->
Result<IggyProducerBuilder, IggyError> {
Ok(IggyProducerBuilder::new(
self.client.clone(),
- stream.try_into()?,
+ Identifier::named(stream)?,
stream.to_owned(),
- topic.try_into()?,
+ Identifier::named(topic)?,
topic.to_owned(),
self.encryptor.clone(),
self.partitioner.clone(),
diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs
index 52a953c4d..1fb152c29 100644
--- a/core/sdk/src/clients/consumer.rs
+++ b/core/sdk/src/clients/consumer.rs
@@ -944,8 +944,9 @@ impl IggyConsumer {
/// [`auto_commit()`](crate::prelude::IggyConsumerBuilder::auto_commit)
is
/// [`AutoCommitWhen::PollingMessages`], checked before anything is
sent. See
/// [Encryption](IggyConsumer#encryption).
- /// - [`IggyError::StreamNameNotFound`] or
[`IggyError::TopicNameNotFound`] when the
- /// stream or the topic still does not exist once the retries are
exhausted.
+ /// - [`IggyError::StreamNameNotFound`], [`IggyError::StreamIdNotFound`]
or [`IggyError::TopicNameNotFound`],
+ /// [`IggyError::TopicIdNotFound`] when the stream or the topic still
does not exist once the retries are
+ /// exhausted.
/// - [`IggyError::ConsumerGroupNameNotFound`] when the consumer group
does not exist
/// and its auto creation is disabled.
/// - Any error returned by the server while looking up the stream or the
topic, or
@@ -1024,17 +1025,26 @@ impl IggyConsumer {
if !stream_exists {
error!("Stream: {stream_id} was not found.");
- return Err(IggyError::StreamNameNotFound(
- self.stream_id.get_string_value().unwrap_or_default(),
- ));
- };
+ return Err(match stream_id.kind {
+ IdKind::String =>
IggyError::StreamNameNotFound(stream_id.get_string_value()?),
+ IdKind::Numeric => {
+
IggyError::StreamIdNotFound(Identifier::from_identifier(&stream_id))
+ }
+ });
+ }
if !topic_exists {
error!("Topic: {topic_id} was not found in stream:
{stream_id}.");
- return Err(IggyError::TopicNameNotFound(
- self.topic_id.get_string_value().unwrap_or_default(),
- self.stream_id.get_string_value().unwrap_or_default(),
- ));
+ return Err(match topic_id.kind {
+ IdKind::String => IggyError::TopicNameNotFound(
+ topic_id.get_string_value()?,
+ stream_id.to_string(),
+ ),
+ IdKind::Numeric => IggyError::TopicIdNotFound(
+ Identifier::from_identifier(&topic_id),
+ Identifier::from_identifier(&stream_id),
+ ),
+ });
}
}
@@ -1156,7 +1166,6 @@ impl IggyConsumer {
self.create_consumer_group_if_not_exists,
self.stream_id.clone(),
self.topic_id.clone(),
- self.consumer.clone(),
&self.consumer_name,
self.joined_consumer_group.clone(),
)
@@ -1257,7 +1266,6 @@ impl IggyConsumer {
create_consumer_group_if_not_exists,
stream_id.clone(),
topic_id.clone(),
- consumer.clone(),
&consumer_name,
joined_consumer_group.clone(),
)
@@ -1420,7 +1428,6 @@ impl IggyConsumer {
create_consumer_group_if_not_exists: bool,
stream_id: Arc<Identifier>,
topic_id: Arc<Identifier>,
- consumer: Arc<Consumer>,
consumer_name: &str,
joined_consumer_group: Arc<AtomicBool>,
) -> Result<(), IggyError> {
@@ -1429,12 +1436,7 @@ impl IggyConsumer {
}
let client = client.read().await;
- let (name, _id) = match consumer.id.kind {
- IdKind::Numeric => (consumer_name.to_owned(),
Some(consumer.id.get_u32_value()?)),
- IdKind::String => (consumer.id.get_string_value()?, None),
- };
-
- let consumer_group_id = name.to_owned().try_into()?;
+ let consumer_group_id = Identifier::named(consumer_name)?;
trace!(
"Validating consumer group: {consumer_group_id} for topic:
{topic_id}, stream: {stream_id}"
);
@@ -1447,7 +1449,7 @@ impl IggyConsumer {
error!("Consumer group does not exist and auto-creation is
disabled.");
let topic_identifier = Identifier::from_identifier(&topic_id);
return Err(IggyError::ConsumerGroupNameNotFound(
- name.to_owned(),
+ consumer_name.to_owned(),
topic_identifier,
));
}
@@ -1456,7 +1458,7 @@ impl IggyConsumer {
"Creating consumer group: {consumer_group_id} for topic:
{topic_id}, stream: {stream_id}"
);
match client
- .create_consumer_group(&stream_id, &topic_id, &name)
+ .create_consumer_group(&stream_id, &topic_id, consumer_name)
.await
{
Ok(_) => {}
diff --git a/core/sdk/src/clients/producer.rs b/core/sdk/src/clients/producer.rs
index d19defcf2..e94cb02ab 100644
--- a/core/sdk/src/clients/producer.rs
+++ b/core/sdk/src/clients/producer.rs
@@ -109,43 +109,51 @@ impl ProducerCore {
let client = self.client.clone();
let client = client.read().await;
if client.get_stream(&stream_id).await?.is_none() {
+ let name = match stream_id.kind {
+ IdKind::String => stream_id.get_string_value()?,
+ IdKind::Numeric => {
+ error!(
+ "Stream: {stream_id} does not exist and cannot be
created from a numeric identifier."
+ );
+ return
Err(IggyError::StreamIdNotFound(Identifier::from_identifier(
+ &stream_id,
+ )));
+ }
+ };
+
if !self.create_stream_if_not_exists {
error!("Stream does not exist and auto-creation is disabled.");
- return
Err(IggyError::StreamNameNotFound(self.stream_name.clone()));
+ return Err(IggyError::StreamNameNotFound(name));
}
- let (name, _id) = match stream_id.kind {
- IdKind::Numeric => (
- self.stream_name.to_owned(),
- Some(self.stream_id.get_u32_value()?),
- ),
- IdKind::String => (self.stream_id.get_string_value()?, None),
- };
info!("Creating stream: {name}");
client.create_stream(&name).await?;
}
if client.get_topic(&stream_id, &topic_id).await?.is_none() {
+ let name = match topic_id.kind {
+ IdKind::String => topic_id.get_string_value()?,
+ IdKind::Numeric => {
+ error!(
+ "Topic: {topic_id} does not exist and cannot be
created from a numeric identifier."
+ );
+ return Err(IggyError::TopicIdNotFound(
+ Identifier::from_identifier(&topic_id),
+ Identifier::from_identifier(&stream_id),
+ ));
+ }
+ };
+
if !self.create_topic_if_not_exists {
error!("Topic does not exist and auto-creation is disabled.");
- return Err(IggyError::TopicNameNotFound(
- self.topic_name.clone(),
- self.stream_name.clone(),
- ));
+ return Err(IggyError::TopicNameNotFound(name,
stream_id.to_string()));
}
- let (name, _id) = match self.topic_id.kind {
- IdKind::Numeric => (
- self.topic_name.to_owned(),
- Some(self.topic_id.get_u32_value()?),
- ),
- IdKind::String => (self.topic_id.get_string_value()?, None),
- };
- info!("Creating topic: {name} for stream: {}", self.stream_name);
+ info!("Creating topic: {name} for stream: {stream_id}");
client
.create_topic(
- &self.stream_id,
- &self.topic_name,
+ &stream_id,
+ &name,
&TopicCreateOptions {
partitions_count: Some(self.topic_partitions_count),
message_expiry: (self.topic_message_expiry !=
IggyExpiry::ServerDefault)
@@ -896,8 +904,9 @@ impl IggyProducer {
///
/// # Errors
///
- /// - [`IggyError::StreamNameNotFound`] or
[`IggyError::TopicNameNotFound`] when the stream or
- /// the topic does not exist and its auto creation is disabled.
+ /// - [`IggyError::StreamNameNotFound`], [`IggyError::StreamIdNotFound`]
or [`IggyError::TopicNameNotFound`],
+ /// [`IggyError::TopicIdNotFound`] when the stream or the topic still
does not exist once the retries are
+ /// exhausted.
/// - Any other error the server raised while looking up or creating the
stream or the topic.
pub async fn init(&self) -> Result<(), IggyError> {
self.core.init().await
diff --git a/core/sdk/src/stream_builder/build/build_stream_topic.rs
b/core/sdk/src/stream_builder/build/build_stream_topic.rs
index beebbcd77..d4a05552f 100644
--- a/core/sdk/src/stream_builder/build/build_stream_topic.rs
+++ b/core/sdk/src/stream_builder/build/build_stream_topic.rs
@@ -17,7 +17,7 @@
*/
use crate::prelude::{
- IdKind, Identifier, IggyClient, IggyError, StreamClient, TopicClient,
TopicCreateOptions,
+ Identifier, IggyClient, IggyError, StreamClient, TopicClient,
TopicCreateOptions,
};
use crate::stream_builder::IggyConsumerConfig;
@@ -43,13 +43,14 @@ pub(crate) async fn build_iggy_stream_topic_if_not_exists(
client: &IggyClient,
config: &IggyConsumerConfig,
) -> Result<(), IggyError> {
- let stream_id = config.stream_id();
let stream_name = config.stream_name();
- let topic_id = config.topic_id();
let topic_name = config.topic_name();
+ let stream_id = Identifier::named(stream_name)?;
+ let topic_id = Identifier::named(topic_name)?;
+
trace!("Check if stream exists.");
- if client.get_stream(config.stream_id()).await?.is_none() {
+ if client.get_stream(&stream_id).await?.is_none() {
trace!("Check if stream should be created.");
if !config.create_stream_if_not_exists() {
warn!(
@@ -59,17 +60,12 @@ pub(crate) async fn build_iggy_stream_topic_if_not_exists(
return Ok(());
}
- let (name, _id) = extract_name_id_from_identifier(stream_id,
stream_name)?;
- trace!("Creating stream: {name}");
- client.create_stream(&name).await?;
+ trace!("Creating stream: {stream_name}");
+ client.create_stream(stream_name).await?;
}
trace!("Check if topic exists.");
- if client
- .get_topic(config.stream_id(), config.topic_id())
- .await?
- .is_none()
- {
+ if client.get_topic(&stream_id, &topic_id).await?.is_none() {
trace!("Check if topic should be created.");
if !config.create_topic_if_not_exists() {
warn!(
@@ -79,18 +75,13 @@ pub(crate) async fn build_iggy_stream_topic_if_not_exists(
return Ok(());
}
- let stream_id = config.stream_id();
- let stream_name = config.stream_name();
- let topic_partitions_count = config.partitions_count();
-
- let (name, _id) = extract_name_id_from_identifier(topic_id,
topic_name)?;
- trace!("Create topic: {name} for stream: {}", stream_name);
+ trace!("Create topic: {topic_name} for stream: {stream_name}");
client
.create_topic(
- stream_id,
+ &stream_id,
topic_name,
&TopicCreateOptions {
- partitions_count: Some(topic_partitions_count),
+ partitions_count: Some(config.partitions_count()),
..TopicCreateOptions::default()
},
)
@@ -99,14 +90,3 @@ pub(crate) async fn build_iggy_stream_topic_if_not_exists(
Ok(())
}
-
-fn extract_name_id_from_identifier(
- stream_id: &Identifier,
- stream_name: &str,
-) -> Result<(String, Option<u32>), IggyError> {
- let (name, id) = match stream_id.kind {
- IdKind::Numeric => (stream_name.to_owned(),
Some(stream_id.get_u32_value()?)),
- IdKind::String => (stream_id.get_string_value()?, None),
- };
- Ok((name, id))
-}
diff --git a/core/sdk/src/stream_builder/config/config_iggy_consumer.rs
b/core/sdk/src/stream_builder/config/config_iggy_consumer.rs
index 83aef80f7..b6da466c6 100644
--- a/core/sdk/src/stream_builder/config/config_iggy_consumer.rs
+++ b/core/sdk/src/stream_builder/config/config_iggy_consumer.rs
@@ -30,12 +30,8 @@ const DEFAULT_PARTITION_ID: u32 = 0;
#[derive(Builder, Debug, Clone)]
#[builder(on(String, into))]
pub struct IggyConsumerConfig {
- /// Identifier of the stream. Must be unique.
- stream_id: Identifier,
/// Name of the stream. Must be unique.
stream_name: String,
- /// Identifier of the topic. Must be unique.
- topic_id: Identifier,
/// Name of the topic. Must be unique.
topic_name: String,
/// The auto-commit configuration for storing the message offset on the
server. See `AutoCommit` for details.
@@ -74,13 +70,8 @@ pub struct IggyConsumerConfig {
impl Default for IggyConsumerConfig {
fn default() -> Self {
- let stream_id = Identifier::from_str_value("test_stream").unwrap();
- let topic_id = Identifier::from_str_value("test_topic").unwrap();
-
Self {
- stream_id,
stream_name: "test_stream".to_string(),
- topic_id,
topic_name: "test_topic".to_string(),
auto_commit: AutoCommit::When(AutoCommitWhen::PollingMessages),
batch_length: 100,
@@ -105,9 +96,7 @@ impl IggyConsumerConfig {
///
/// # Args
///
- /// * `stream_id` - The stream id.
/// * `stream_name` - The stream name.
- /// * `topic_id` - The topic id.
/// * `topic_name` - The topic name.
/// * `auto_commit` - The auto commit config.
/// * `batch_length` - The max number of messages to poll in a batch.
@@ -131,9 +120,7 @@ impl IggyConsumerConfig {
///
#[allow(clippy::too_many_arguments)]
pub fn new(
- stream_id: Identifier,
stream_name: String,
- topic_id: Identifier,
topic_name: String,
auto_commit: AutoCommit,
batch_length: u32,
@@ -150,9 +137,7 @@ impl IggyConsumerConfig {
init_interval: NonZeroIggyDuration,
) -> Self {
Self {
- stream_id,
stream_name,
- topic_id,
topic_name,
auto_commit,
batch_length,
@@ -189,13 +174,12 @@ impl IggyConsumerConfig {
batch_length: u32,
polling_interval: IggyDuration,
) -> Result<Self, IggyError> {
- let stream_id = Identifier::from_str_value(stream)?;
- let topic_id = Identifier::from_str_value(topic)?;
+ // Validate stream and topic names. Returns IggyError if names are not
allowed.
+ Identifier::named(stream)?;
+ Identifier::named(topic)?;
Ok(Self {
- stream_id,
stream_name: stream.to_string(),
- topic_id,
topic_name: topic.to_string(),
auto_commit: AutoCommit::When(AutoCommitWhen::PollingMessages),
batch_length,
@@ -222,18 +206,10 @@ impl IggyConsumerConfig {
self
}
- pub fn stream_id(&self) -> &Identifier {
- &self.stream_id
- }
-
pub fn stream_name(&self) -> &str {
&self.stream_name
}
- pub fn topic_id(&self) -> &Identifier {
- &self.topic_id
- }
-
pub fn topic_name(&self) -> &str {
&self.topic_name
}
@@ -335,14 +311,9 @@ mod tests {
#[test]
fn should_be_equal() {
- let stream_id = Identifier::from_str_value("test_stream").unwrap();
- let topic_id = Identifier::from_str_value("test_topic").unwrap();
-
// Builder is generated by the bon macro
let config = IggyConsumerConfig::builder()
- .stream_id(stream_id)
.stream_name("test_stream".to_string())
- .topic_id(topic_id)
.topic_name("test_topic".to_string())
.auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages))
.batch_length(100)
@@ -358,15 +329,7 @@ mod tests {
.init_interval(NonZeroIggyDuration::from_str("3s").unwrap())
.build();
- assert_eq!(
- config.stream_id(),
- &Identifier::from_str_value("test_stream").unwrap()
- );
assert_eq!(config.stream_name(), "test_stream");
- assert_eq!(
- config.topic_id(),
- &Identifier::from_str_value("test_topic").unwrap()
- );
assert_eq!(config.topic_name(), "test_topic");
assert_eq!(
config.auto_commit(),
@@ -398,13 +361,8 @@ mod tests {
#[test]
fn should_be_default() {
- let stream_id = Identifier::from_str_value("test_stream").unwrap();
- let topic_id = Identifier::from_str_value("test_topic").unwrap();
-
let config = IggyConsumerConfig::default();
- assert_eq!(config.stream_id(), &stream_id);
assert_eq!(config.stream_name(), "test_stream");
- assert_eq!(config.topic_id(), &topic_id);
assert_eq!(config.topic_name(), "test_topic");
assert_eq!(
config.auto_commit(),
@@ -436,9 +394,7 @@ mod tests {
#[test]
fn should_be_new() {
let config = IggyConsumerConfig::new(
- Identifier::from_str_value("test_stream").unwrap(),
"test_stream".to_string(),
- Identifier::from_str_value("test_topic").unwrap(),
"test_topic".to_string(),
AutoCommit::When(AutoCommitWhen::PollingMessages),
100,
@@ -454,15 +410,7 @@ mod tests {
Some(3),
NonZeroIggyDuration::from_str("3s").unwrap(),
);
- assert_eq!(
- config.stream_id(),
- &Identifier::from_str_value("test_stream").unwrap(),
- );
assert_eq!(config.stream_name(), "test_stream");
- assert_eq!(
- config.topic_id(),
- &Identifier::from_str_value("test_topic").unwrap()
- );
assert_eq!(config.topic_name(), "test_topic");
assert_eq!(
config.auto_commit(),
diff --git a/core/sdk/src/stream_builder/config/config_iggy_producer.rs
b/core/sdk/src/stream_builder/config/config_iggy_producer.rs
index bfcc7f2d4..9acd2a80a 100644
--- a/core/sdk/src/stream_builder/config/config_iggy_producer.rs
+++ b/core/sdk/src/stream_builder/config/config_iggy_producer.rs
@@ -26,12 +26,8 @@ use std::sync::Arc;
#[derive(Builder, Debug, Clone)]
#[builder(on(String, into))]
pub struct IggyProducerConfig {
- /// Identifier of the stream. Must be unique.
- stream_id: Identifier,
/// Name of the stream. Must be unique.
stream_name: String,
- /// Identifier of the topic. Must be unique.
- topic_id: Identifier,
/// Name of the topic. Must be unique.
topic_name: String,
/// Sets the number of partitions to create for the topic
@@ -57,13 +53,8 @@ pub struct IggyProducerConfig {
impl Default for IggyProducerConfig {
fn default() -> Self {
- let stream_id = Identifier::from_str_value("test_stream").unwrap();
- let topic_id = Identifier::from_str_value("test_topic").unwrap();
-
Self {
- stream_id,
stream_name: "test_stream".to_string(),
- topic_id,
topic_name: "test_topic".to_string(),
batch_length: 100,
linger_time: IggyDuration::from_str("5ms").unwrap(),
@@ -82,9 +73,7 @@ impl IggyProducerConfig {
///
/// # Args
///
- /// * `stream_id` - The stream identifier.
/// * `stream_name` - The stream name.
- /// * `topic_id` - The topic identifier.
/// * `topic_name` - The topic name.
/// * `topic_partitions_count` - The number of partitions to create.
/// * `batch_length` - The max number of messages to send in a batch.
@@ -99,9 +88,7 @@ impl IggyProducerConfig {
///
#[allow(clippy::too_many_arguments)]
pub fn new(
- stream_id: Identifier,
stream_name: String,
- topic_id: Identifier,
topic_name: String,
topic_partitions_count: u32,
batch_length: u32,
@@ -112,9 +99,7 @@ impl IggyProducerConfig {
send_retries_interval: Option<NonZeroIggyDuration>,
) -> Self {
Self {
- stream_id,
stream_name,
- topic_id,
topic_name,
topic_partitions_count,
topic_durability: Durability::default(),
@@ -146,13 +131,12 @@ impl IggyProducerConfig {
batch_length: u32,
linger_time: IggyDuration,
) -> Result<Self, IggyError> {
- let stream_id = Identifier::from_str_value(stream)?;
- let topic_id = Identifier::from_str_value(topic)?;
+ // Validate stream and topic names. Returns IggyError if names are not
allowed.
+ Identifier::named(stream)?;
+ Identifier::named(topic)?;
Ok(Self {
- stream_id,
stream_name: stream.to_string(),
- topic_id,
topic_name: topic.to_string(),
batch_length,
linger_time,
@@ -167,18 +151,10 @@ impl IggyProducerConfig {
}
impl IggyProducerConfig {
- pub fn stream_id(&self) -> &Identifier {
- &self.stream_id
- }
-
pub fn stream_name(&self) -> &str {
&self.stream_name
}
- pub fn topic_id(&self) -> &Identifier {
- &self.topic_id
- }
-
pub fn topic_name(&self) -> &str {
&self.topic_name
}
@@ -233,9 +209,7 @@ mod tests {
// Builder is generated by the bon macro
let config = IggyProducerConfig::builder()
- .stream_id(Identifier::from_str_value(stream).unwrap())
.stream_name(stream)
- .topic_id(Identifier::from_str_value(topic).unwrap())
.topic_name(topic)
.topic_partitions_count(3)
.batch_length(100)
@@ -245,15 +219,7 @@ mod tests {
.send_retries_interval(NonZeroIggyDuration::ONE_SECOND)
.build();
- assert_eq!(
- config.stream_id(),
- &Identifier::from_str_value("test_stream").unwrap()
- );
assert_eq!(config.stream_name(), "test_stream");
- assert_eq!(
- config.topic_id(),
- &Identifier::from_str_value("test_topic").unwrap()
- );
assert_eq!(config.topic_name(), "test_topic");
assert_eq!(config.batch_length(), 100);
assert_eq!(config.linger_time(),
IggyDuration::from_str("5ms").unwrap());
@@ -268,13 +234,8 @@ mod tests {
#[test]
fn should_be_default() {
- let stream_id = Identifier::from_str_value("test_stream").unwrap();
- let topic_id = Identifier::from_str_value("test_topic").unwrap();
-
let config = IggyProducerConfig::default();
- assert_eq!(config.stream_id(), &stream_id);
assert_eq!(config.stream_name(), "test_stream");
- assert_eq!(config.topic_id(), &topic_id);
assert_eq!(config.topic_name(), "test_topic");
assert_eq!(config.batch_length(), 100);
assert_eq!(config.linger_time(),
IggyDuration::from_str("5ms").unwrap());
@@ -289,13 +250,8 @@ mod tests {
#[test]
fn should_be_new() {
- let stream_id = Identifier::from_str_value("test_stream").unwrap();
- let topic_id = Identifier::from_str_value("test_topic").unwrap();
-
let config = IggyProducerConfig::new(
- stream_id.clone(),
String::from("test_stream"),
- topic_id.clone(),
String::from("test_topic"),
3,
100,
@@ -305,9 +261,7 @@ mod tests {
None,
None,
);
- assert_eq!(config.stream_id(), &stream_id);
assert_eq!(config.stream_name(), "test_stream");
- assert_eq!(config.topic_id(), &topic_id);
assert_eq!(config.topic_name(), "test_topic");
assert_eq!(config.batch_length(), 100);
assert_eq!(config.linger_time(),
IggyDuration::from_str("5ms").unwrap());
@@ -319,9 +273,6 @@ mod tests {
#[test]
fn should_be_from_stream_topic() {
- let stream_id = Identifier::from_str_value("test_stream").unwrap();
- let topic_id = Identifier::from_str_value("test_topic").unwrap();
-
let res = IggyProducerConfig::from_stream_topic(
"test_stream",
"test_topic",
@@ -332,9 +283,7 @@ mod tests {
assert!(res.is_ok());
let config = res.unwrap();
- assert_eq!(config.stream_id(), &stream_id);
assert_eq!(config.stream_name(), "test_stream");
- assert_eq!(config.topic_id(), &topic_id);
assert_eq!(config.topic_name(), "test_topic");
assert_eq!(config.batch_length(), 100);
assert_eq!(config.linger_time(),
IggyDuration::from_str("5ms").unwrap());
diff --git a/core/sdk/src/stream_builder/config/config_iggy_stream.rs
b/core/sdk/src/stream_builder/config/config_iggy_stream.rs
index 45792e2ba..b3277e76c 100644
--- a/core/sdk/src/stream_builder/config/config_iggy_stream.rs
+++ b/core/sdk/src/stream_builder/config/config_iggy_stream.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::prelude::{Identifier, IggyDuration, IggyError};
+use crate::prelude::{IggyDuration, IggyError};
use crate::stream_builder::{IggyConsumerConfig, IggyProducerConfig};
use bon::Builder;
@@ -75,9 +75,7 @@ impl IggyStreamConfig {
producer_config,
})
}
-}
-impl IggyStreamConfig {
pub fn consumer_config(&self) -> &IggyConsumerConfig {
&self.consumer_config
}
@@ -85,22 +83,6 @@ impl IggyStreamConfig {
pub fn producer_config(&self) -> &IggyProducerConfig {
&self.producer_config
}
-
- pub fn stream_id(&self) -> &Identifier {
- self.producer_config.stream_id()
- }
-
- pub fn stream_name(&self) -> &str {
- self.producer_config.stream_name()
- }
-
- pub fn topic_id(&self) -> &Identifier {
- self.producer_config.topic_id()
- }
-
- pub fn topic_name(&self) -> &str {
- self.producer_config.topic_name()
- }
}
#[cfg(test)]
@@ -126,8 +108,8 @@ mod tests {
)
.unwrap();
let config = IggyStreamConfig::new(consumer_config, producer_config);
- assert_eq!(config.stream_name(), "test_stream");
- assert_eq!(config.topic_name(), "test_topic");
+ assert_eq!(config.consumer_config().stream_name(), "test_stream");
+ assert_eq!(config.producer_config().topic_name(), "test_topic");
assert_eq!(config.consumer_config().batch_length(), 100);
assert_eq!(config.producer_config().batch_length(), 100);
assert_eq!(
@@ -143,8 +125,8 @@ mod tests {
#[test]
fn should_be_default() {
let config = IggyStreamConfig::default();
- assert_eq!(config.stream_name(), "test_stream");
- assert_eq!(config.topic_name(), "test_topic");
+ assert_eq!(config.consumer_config().stream_name(), "test_stream");
+ assert_eq!(config.producer_config().topic_name(), "test_topic");
assert_eq!(config.consumer_config().batch_length(), 100);
assert_eq!(config.producer_config().batch_length(), 100);
assert_eq!(
@@ -170,8 +152,8 @@ mod tests {
assert!(res.is_ok());
let config = res.unwrap();
- assert_eq!(config.stream_name(), "test_stream");
- assert_eq!(config.topic_name(), "test_topic");
+ assert_eq!(config.consumer_config().stream_name(), "test_stream");
+ assert_eq!(config.producer_config().topic_name(), "test_topic");
assert_eq!(config.consumer_config().batch_length(), 100);
assert_eq!(config.producer_config().batch_length(), 100);
assert_eq!(
diff --git a/examples/rust/src/shared/system.rs
b/examples/rust/src/shared/system.rs
index 6f4358e91..c90d2002d 100644
--- a/examples/rust/src/shared/system.rs
+++ b/examples/rust/src/shared/system.rs
@@ -28,8 +28,8 @@ pub async fn init_by_consumer(args: &Args, client: &dyn
Client) {
args.partition_id,
);
let mut interval =
tokio::time::interval(std::time::Duration::from_secs(1));
- let stream_id = stream_id.try_into().unwrap();
- let topic_id = topic_id.try_into().unwrap();
+ let stream_id = Identifier::named(&stream_id).unwrap();
+ let topic_id = Identifier::named(&topic_id).unwrap();
loop {
interval.tick().await;
info!("Validating if stream: {stream_id} exists..");
@@ -73,7 +73,7 @@ pub async fn init_by_consumer(args: &Args, client: &dyn
Client) {
}
pub async fn init_by_producer(args: &Args, client: &dyn Client) -> Result<(),
IggyError> {
- let stream_id = args.stream_id.clone().try_into()?;
+ let stream_id = Identifier::named(&args.stream_id)?;
let topic_name = args.topic_id.clone();
let stream = client.get_stream(&stream_id).await?;
if stream.is_some() {
@@ -114,8 +114,8 @@ pub async fn consume_messages(
interval.map_or("none".to_string(), |i| i.as_human_time_string())
);
- let stream_id = args.stream_id.clone().try_into()?;
- let topic_id = args.topic_id.clone().try_into()?;
+ let stream_id = Identifier::named(&args.stream_id)?;
+ let topic_id = Identifier::named(&args.topic_id)?;
let mut interval = interval.map(|interval|
tokio::time::interval(interval.get_duration()));
let mut consumed_batches = 0;
let consumer = Consumer {
diff --git a/examples/rust/src/stream-builder/stream-basic/main.rs
b/examples/rust/src/stream-builder/stream-basic/main.rs
index 70edcf49b..08d77048c 100644
--- a/examples/rust/src/stream-builder/stream-basic/main.rs
+++ b/examples/rust/src/stream-builder/stream-basic/main.rs
@@ -61,7 +61,8 @@ async fn main() -> Result<(), IggyError> {
println!("Stop the message stream and shutdown iggy client");
sender.send(()).expect("Failed to send shutdown signal");
- client.delete_stream(stream_config.stream_id()).await?;
+ let stream_id =
Identifier::named(stream_config.producer_config().stream_name())?;
+ client.delete_stream(&stream_id).await?;
client.shutdown().await?;
Ok(())
diff --git a/examples/rust/src/stream-builder/stream-consumer-config/main.rs
b/examples/rust/src/stream-builder/stream-consumer-config/main.rs
index ea02ea3db..874c197f0 100644
--- a/examples/rust/src/stream-builder/stream-consumer-config/main.rs
+++ b/examples/rust/src/stream-builder/stream-consumer-config/main.rs
@@ -28,11 +28,9 @@ async fn main() -> Result<(), IggyError> {
let topic = "test_topic";
let config = IggyConsumerConfig::builder()
- // Set the stream identifier and name.
- .stream_id(Identifier::from_str_value(stream)?)
+ // Set the stream name. The consumer binds to the stream by this name.
.stream_name(stream)
- // Set the topic identifier and name
- .topic_id(Identifier::from_str_value(topic)?)
+ // Set the topic name. The consumer binds to the topic by this name.
.topic_name(topic)
// The auto-commit configuration for storing the message offset on the
server.
// * Disabled: The auto-commit is disabled and the offset must be
stored manually by the consumer.
diff --git a/examples/rust/src/stream-builder/stream-producer-config/main.rs
b/examples/rust/src/stream-builder/stream-producer-config/main.rs
index 00afb7bca..5a34cb08b 100644
--- a/examples/rust/src/stream-builder/stream-producer-config/main.rs
+++ b/examples/rust/src/stream-builder/stream-producer-config/main.rs
@@ -27,11 +27,9 @@ async fn main() -> Result<(), IggyError> {
// The builder simplifies the IggyProducer configuration.
let config = IggyProducerConfig::builder()
- // Set the stream identifier and name.
- .stream_id(Identifier::from_str_value(stream)?)
+ // Set the stream name. The producer binds to the stream by this name.
.stream_name(stream)
- // Set the topic identifier and name
- .topic_id(Identifier::from_str_value(topic)?)
+ // Set the topic name. The producer binds to the topic by this name.
.topic_name(topic)
// Sets the number of partitions to create for the topic.
// The more clients are reading concurrently, the more partitions you
should create.
