You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/09/23 12:00:00 UTC
(iggy) branch io_uring_tpc updated: feat(io_uring): remove old mod cluster, reformat code (#2194)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/io_uring_tpc by this push:
new 4309609e feat(io_uring): remove old mod cluster, reformat code (#2194)
4309609e is described below
commit 4309609e6eb6e2695433618ff8b9e89b25c85bd3
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Tue Sep 23 13:59:57 2025 +0200
feat(io_uring): remove old mod cluster, reformat code (#2194)
---
core/bench/src/benchmarks/common.rs | 6 +-
.../create_consumer_group.rs | 6 +-
.../src/cli/binary_topics/create_topic.rs | 14 ++++-
.../src/client/binary_clients/stream_client.rs | 5 +-
.../src/client/binary_streams/mod.rs | 5 +-
core/cli/src/main.rs | 4 +-
.../consumer_groups/create_consumer_group.rs | 8 +--
.../test_consumer_group_create_command.rs | 4 +-
.../test_consumer_group_delete_command.rs | 4 +-
.../test_consumer_group_get_command.rs | 4 +-
.../test_consumer_group_list_command.rs | 8 +--
.../test_consumer_offset_get_command.rs | 4 +-
.../test_consumer_offset_set_command.rs | 4 +-
.../cli/message/test_message_flush_command.rs | 16 +++--
.../tests/cli/message/test_message_poll_command.rs | 2 +-
.../tests/cli/message/test_message_send_command.rs | 10 ++-
.../tests/cli/stream/test_stream_purge_command.rs | 4 +-
.../tests/cli/topic/test_topic_create_command.rs | 4 +-
.../tests/cli/topic/test_topic_delete_command.rs | 4 +-
.../tests/cli/topic/test_topic_get_command.rs | 4 +-
.../tests/cli/topic/test_topic_list_command.rs | 4 +-
.../tests/cli/topic/test_topic_purge_command.rs | 4 +-
.../tests/cli/topic/test_topic_update_command.rs | 4 +-
core/integration/tests/sdk/producer/mod.rs | 5 +-
.../scenarios/consumer_group_join_scenario.rs | 29 ++++-----
...h_multiple_clients_polling_messages_scenario.rs | 10 +--
...with_single_client_polling_messages_scenario.rs | 10 +--
.../server/scenarios/create_message_payload.rs | 5 +-
.../server/scenarios/delete_segments_scenario.rs | 5 +-
.../tests/server/scenarios/encryption_scenario.rs | 8 +--
.../server/scenarios/message_headers_scenario.rs | 8 +--
.../server/scenarios/message_size_scenario.rs | 7 +--
.../tests/server/scenarios/system_scenario.rs | 14 ++---
.../tests/server/scenarios/tcp_tls_scenario.rs | 5 +-
core/integration/tests/state/system.rs | 15 +----
.../src/client_wrappers/binary_stream_client.rs | 5 +-
core/sdk/src/clients/binary_streams.rs | 11 +---
core/sdk/src/clients/consumer.rs | 3 +-
core/sdk/src/http/binary_streams.rs | 5 +-
.../handlers/streams/create_stream_handler.rs | 4 +-
core/server/src/binary/mapper.rs | 6 +-
.../commands/clean_personal_access_tokens.rs | 2 +-
core/server/src/channels/commands/print_sysinfo.rs | 2 +-
core/server/src/channels/commands/save_messages.rs | 6 +-
core/server/src/http/system.rs | 3 +-
core/server/src/quic/listener.rs | 3 +-
core/server/src/shard/system/clients.rs | 4 +-
core/server/src/shard/system/cluster.rs | 6 +-
core/server/src/shard/transmission/event.rs | 3 +-
.../server/src/streaming/clients/client_manager.rs | 6 +-
core/server/src/streaming/cluster/mod.rs | 71 ----------------------
core/server/src/streaming/mod.rs | 1 -
core/tools/src/data-seeder/seeder.rs | 8 +--
examples/rust/src/getting-started/producer/main.rs | 24 +++++---
54 files changed, 138 insertions(+), 288 deletions(-)
diff --git a/core/bench/src/benchmarks/common.rs b/core/bench/src/benchmarks/common.rs
index 6b588646..3ee4f686 100644
--- a/core/bench/src/benchmarks/common.rs
+++ b/core/bench/src/benchmarks/common.rs
@@ -88,11 +88,7 @@ pub async fn init_consumer_groups(
consumer_group_name, stream_name, topic_id
);
match client
- .create_consumer_group(
- &stream_id,
- &topic_id,
- &consumer_group_name,
- )
+ .create_consumer_group(&stream_id, &topic_id, &consumer_group_name)
.await
{
Err(IggyError::ConsumerGroupIdAlreadyExists(_, _)) => {
diff --git a/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs b/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs
index 34e2106a..e8fb0d87 100644
--- a/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs
+++ b/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs
@@ -29,11 +29,7 @@ pub struct CreateConsumerGroupCmd {
}
impl CreateConsumerGroupCmd {
- pub fn new(
- stream_id: Identifier,
- topic_id: Identifier,
- name: String,
- ) -> Self {
+ pub fn new(stream_id: Identifier, topic_id: Identifier, name: String) -> Self {
Self {
create_consumer_group: CreateConsumerGroup {
stream_id,
diff --git a/core/binary_protocol/src/cli/binary_topics/create_topic.rs b/core/binary_protocol/src/cli/binary_topics/create_topic.rs
index d605705e..ff4ef9fd 100644
--- a/core/binary_protocol/src/cli/binary_topics/create_topic.rs
+++ b/core/binary_protocol/src/cli/binary_topics/create_topic.rs
@@ -68,12 +68,22 @@ impl CliCommand for CreateTopicCmd {
async fn execute_cmd(&mut self, client: &dyn Client) -> anyhow::Result<(), anyhow::Error> {
client
- .create_topic(&self.create_topic.stream_id, &self.create_topic.name, self.create_topic.partitions_count, self.create_topic.compression_algorithm, self.create_topic.replication_factor, self.create_topic.message_expiry, self.create_topic.max_topic_size)
+ .create_topic(
+ &self.create_topic.stream_id,
+ &self.create_topic.name,
+ self.create_topic.partitions_count,
+ self.create_topic.compression_algorithm,
+ self.create_topic.replication_factor,
+ self.create_topic.message_expiry,
+ self.create_topic.max_topic_size,
+ )
.await
.with_context(|| {
format!(
"Problem creating topic (name: {}, partitions count: {}) in stream with ID: {}",
- self.create_topic.name, self.create_topic.partitions_count, self.create_topic.stream_id
+ self.create_topic.name,
+ self.create_topic.partitions_count,
+ self.create_topic.stream_id
)
})?;
diff --git a/core/binary_protocol/src/client/binary_clients/stream_client.rs b/core/binary_protocol/src/client/binary_clients/stream_client.rs
index 490499dd..b56c44b3 100644
--- a/core/binary_protocol/src/client/binary_clients/stream_client.rs
+++ b/core/binary_protocol/src/client/binary_clients/stream_client.rs
@@ -33,10 +33,7 @@ pub trait StreamClient {
/// Create a new stream.
///
/// Authentication is required, and the permission to manage the streams.
- async fn create_stream(
- &self,
- name: &str,
- ) -> Result<StreamDetails, IggyError>;
+ async fn create_stream(&self, name: &str) -> Result<StreamDetails, IggyError>;
/// Update a stream by unique ID or name.
///
/// Authentication is required, and the permission to manage the streams.
diff --git a/core/binary_protocol/src/client/binary_streams/mod.rs b/core/binary_protocol/src/client/binary_streams/mod.rs
index df93fcac..ebb9ab6c 100644
--- a/core/binary_protocol/src/client/binary_streams/mod.rs
+++ b/core/binary_protocol/src/client/binary_streams/mod.rs
@@ -49,10 +49,7 @@ impl<B: BinaryClient> StreamClient for B {
mapper::map_streams(response)
}
- async fn create_stream(
- &self,
- name: &str,
- ) -> Result<StreamDetails, IggyError> {
+ async fn create_stream(&self, name: &str) -> Result<StreamDetails, IggyError> {
fail_if_not_authenticated(self).await?;
let response = self
.send_with_response(&CreateStream {
diff --git a/core/cli/src/main.rs b/core/cli/src/main.rs
index 8451b0d6..21b60aaa 100644
--- a/core/cli/src/main.rs
+++ b/core/cli/src/main.rs
@@ -106,9 +106,7 @@ fn get_command(
#[warn(clippy::let_and_return)]
match command {
Command::Stream(command) => match command {
- StreamAction::Create(args) => {
- Box::new(CreateStreamCmd::new(args.name.clone()))
- }
+ StreamAction::Create(args) => Box::new(CreateStreamCmd::new(args.name.clone())),
StreamAction::Delete(args) => Box::new(DeleteStreamCmd::new(args.stream_id.clone())),
StreamAction::Update(args) => Box::new(UpdateStreamCmd::new(
args.stream_id.clone(),
diff --git a/core/common/src/commands/consumer_groups/create_consumer_group.rs b/core/common/src/commands/consumer_groups/create_consumer_group.rs
index fb11a7c2..c5bd3e91 100644
--- a/core/common/src/commands/consumer_groups/create_consumer_group.rs
+++ b/core/common/src/commands/consumer_groups/create_consumer_group.rs
@@ -111,13 +111,7 @@ impl BytesSerializable for CreateConsumerGroup {
impl Display for CreateConsumerGroup {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- write!(
- f,
- "{}|{}|{}",
- self.stream_id,
- self.topic_id,
- self.name
- )
+ write!(f, "{}|{}|{}", self.stream_id, self.topic_id, self.name)
}
}
diff --git a/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs b/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
index 56bcef2c..b3e716b0 100644
--- a/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
+++ b/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
@@ -86,9 +86,7 @@ impl TestConsumerGroupCreateCmd {
#[async_trait]
impl IggyCmdTestCase for TestConsumerGroupCreateCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let topic = client
diff --git a/core/integration/tests/cli/consumer_group/test_consumer_group_delete_command.rs b/core/integration/tests/cli/consumer_group/test_consumer_group_delete_command.rs
index df9c5b09..1a419177 100644
--- a/core/integration/tests/cli/consumer_group/test_consumer_group_delete_command.rs
+++ b/core/integration/tests/cli/consumer_group/test_consumer_group_delete_command.rs
@@ -89,9 +89,7 @@ impl TestConsumerGroupDeleteCmd {
#[async_trait]
impl IggyCmdTestCase for TestConsumerGroupDeleteCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let topic = client
diff --git a/core/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs b/core/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs
index 5d0ebdd9..c41231aa 100644
--- a/core/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs
+++ b/core/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs
@@ -89,9 +89,7 @@ impl TestConsumerGroupGetCmd {
#[async_trait]
impl IggyCmdTestCase for TestConsumerGroupGetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let topic = client
diff --git a/core/integration/tests/cli/consumer_group/test_consumer_group_list_command.rs b/core/integration/tests/cli/consumer_group/test_consumer_group_list_command.rs
index 7f66937a..0267b378 100644
--- a/core/integration/tests/cli/consumer_group/test_consumer_group_list_command.rs
+++ b/core/integration/tests/cli/consumer_group/test_consumer_group_list_command.rs
@@ -59,7 +59,7 @@ impl TestConsumerGroupListCmd {
fn to_args(&self) -> Vec<String> {
let mut command = vec![];
-
+
// Use actual stream ID if available, otherwise use stream name as fallback
if let Some(stream_id) = self.actual_stream_id {
command.push(format!("{}", stream_id));
@@ -90,7 +90,7 @@ impl IggyCmdTestCase for TestConsumerGroupListCmd {
.expect("Failed to create stream");
self.actual_stream_id = Some(stream.id);
- // Create topic and capture its actual ID
+ // Create topic and capture its actual ID
let topic = client
.create_topic(
&stream.id.try_into().unwrap(),
@@ -173,9 +173,7 @@ impl IggyCmdTestCase for TestConsumerGroupListCmd {
}
if let Some(stream_id) = self.actual_stream_id {
- let stream = client
- .delete_stream(&stream_id.try_into().unwrap())
- .await;
+ let stream = client.delete_stream(&stream_id.try_into().unwrap()).await;
assert!(stream.is_ok());
}
}
diff --git a/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs b/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs
index f15b765e..f7b9dd3a 100644
--- a/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs
+++ b/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs
@@ -201,9 +201,7 @@ impl IggyCmdTestCase for TestConsumerOffsetGetCmd {
}
if let Some(stream_id) = self.actual_stream_id {
- let stream = client
- .delete_stream(&stream_id.try_into().unwrap())
- .await;
+ let stream = client.delete_stream(&stream_id.try_into().unwrap()).await;
assert!(stream.is_ok());
}
}
diff --git a/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs b/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs
index 4a003d22..28328938 100644
--- a/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs
+++ b/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs
@@ -97,9 +97,7 @@ impl TestConsumerOffsetSetCmd {
#[async_trait]
impl IggyCmdTestCase for TestConsumerOffsetSetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let topic = client
diff --git a/core/integration/tests/cli/message/test_message_flush_command.rs b/core/integration/tests/cli/message/test_message_flush_command.rs
index 7dc295f2..3e2b2486 100644
--- a/core/integration/tests/cli/message/test_message_flush_command.rs
+++ b/core/integration/tests/cli/message/test_message_flush_command.rs
@@ -90,7 +90,9 @@ impl TestMessageFetchCmd {
impl IggyCmdTestCase for TestMessageFetchCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
// Create stream and capture its actual ID
- let stream = client.create_stream(&self.stream_name).await
+ let stream = client
+ .create_stream(&self.stream_name)
+ .await
.expect("Failed to create stream");
self.actual_stream_id = Some(stream.id);
@@ -154,7 +156,9 @@ impl IggyCmdTestCase for TestMessageFetchCmd {
assert!(stream_id.is_ok());
let stream_id = stream_id.unwrap();
- let topic = client.delete_topic(&stream_id, &topic_id.try_into().unwrap()).await;
+ let topic = client
+ .delete_topic(&stream_id, &topic_id.try_into().unwrap())
+ .await;
assert!(topic.is_ok());
let stream = client.delete_stream(&stream_id).await;
@@ -173,13 +177,7 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
for fsync in test_parameters {
iggy_cmd_test
- .execute_test(TestMessageFetchCmd::new(
- "stream",
- "topic",
- 1,
- 1,
- fsync,
- ))
+ .execute_test(TestMessageFetchCmd::new("stream", "topic", 1, 1, fsync))
.await;
}
}
diff --git a/core/integration/tests/cli/message/test_message_poll_command.rs b/core/integration/tests/cli/message/test_message_poll_command.rs
index 3227beff..20f5bcf1 100644
--- a/core/integration/tests/cli/message/test_message_poll_command.rs
+++ b/core/integration/tests/cli/message/test_message_poll_command.rs
@@ -238,7 +238,7 @@ impl IggyCmdTestCase for TestMessagePollCmd {
async fn verify_server_state(&self, client: &dyn Client) {
let stream_id = Identifier::from_str(&self.stream_name).unwrap();
let topic_id = Identifier::from_str(&self.topic_name).unwrap();
-
+
let topic = client.delete_topic(&stream_id, &topic_id).await;
assert!(topic.is_ok());
diff --git a/core/integration/tests/cli/message/test_message_send_command.rs b/core/integration/tests/cli/message/test_message_send_command.rs
index 70155458..210daa32 100644
--- a/core/integration/tests/cli/message/test_message_send_command.rs
+++ b/core/integration/tests/cli/message/test_message_send_command.rs
@@ -274,8 +274,14 @@ pub async fn should_be_successful() {
(ProvideMessages::ViaStdin, PartitionSelection::Balanced),
(ProvideMessages::ViaStdin, PartitionSelection::Id(1)),
(ProvideMessages::AsArgs, PartitionSelection::Id(2)),
- (ProvideMessages::AsArgs, PartitionSelection::Key(String::from("some-complex-key"))),
- (ProvideMessages::ViaStdin, PartitionSelection::Key(String::from("another-key"))),
+ (
+ ProvideMessages::AsArgs,
+ PartitionSelection::Key(String::from("some-complex-key")),
+ ),
+ (
+ ProvideMessages::ViaStdin,
+ PartitionSelection::Key(String::from("another-key")),
+ ),
];
iggy_cmd_test.setup().await;
diff --git a/core/integration/tests/cli/stream/test_stream_purge_command.rs b/core/integration/tests/cli/stream/test_stream_purge_command.rs
index 10358da7..d605f8f8 100644
--- a/core/integration/tests/cli/stream/test_stream_purge_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_purge_command.rs
@@ -57,9 +57,7 @@ impl TestStreamPurgeCmd {
#[async_trait]
impl IggyCmdTestCase for TestStreamPurgeCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let topic = client
diff --git a/core/integration/tests/cli/topic/test_topic_create_command.rs b/core/integration/tests/cli/topic/test_topic_create_command.rs
index e485bc68..6cb045c3 100644
--- a/core/integration/tests/cli/topic/test_topic_create_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_create_command.rs
@@ -101,9 +101,7 @@ impl TestTopicCreateCmd {
#[async_trait]
impl IggyCmdTestCase for TestTopicCreateCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
}
diff --git a/core/integration/tests/cli/topic/test_topic_delete_command.rs b/core/integration/tests/cli/topic/test_topic_delete_command.rs
index 3c4f0edc..9f3e5c0e 100644
--- a/core/integration/tests/cli/topic/test_topic_delete_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_delete_command.rs
@@ -74,9 +74,7 @@ impl TestTopicDeleteCmd {
#[async_trait]
impl IggyCmdTestCase for TestTopicDeleteCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let topic = client
diff --git a/core/integration/tests/cli/topic/test_topic_get_command.rs b/core/integration/tests/cli/topic/test_topic_get_command.rs
index 45b0fd01..0fec1f66 100644
--- a/core/integration/tests/cli/topic/test_topic_get_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_get_command.rs
@@ -74,9 +74,7 @@ impl TestTopicGetCmd {
#[async_trait]
impl IggyCmdTestCase for TestTopicGetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let topic = client
diff --git a/core/integration/tests/cli/topic/test_topic_list_command.rs b/core/integration/tests/cli/topic/test_topic_list_command.rs
index 0b91ca58..b025308e 100644
--- a/core/integration/tests/cli/topic/test_topic_list_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_list_command.rs
@@ -71,9 +71,7 @@ impl TestTopicListCmd {
#[async_trait]
impl IggyCmdTestCase for TestTopicListCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let topic = client
diff --git a/core/integration/tests/cli/topic/test_topic_purge_command.rs b/core/integration/tests/cli/topic/test_topic_purge_command.rs
index ae310a7f..ee39cbe6 100644
--- a/core/integration/tests/cli/topic/test_topic_purge_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_purge_command.rs
@@ -73,9 +73,7 @@ impl TestTopicPurgeCmd {
#[async_trait]
impl IggyCmdTestCase for TestTopicPurgeCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let topic = client
diff --git a/core/integration/tests/cli/topic/test_topic_update_command.rs b/core/integration/tests/cli/topic/test_topic_update_command.rs
index 9ed19bd5..f37e957e 100644
--- a/core/integration/tests/cli/topic/test_topic_update_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_update_command.rs
@@ -122,9 +122,7 @@ impl TestTopicUpdateCmd {
#[async_trait]
impl IggyCmdTestCase for TestTopicUpdateCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name)
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let message_expiry = match &self.message_expiry {
diff --git a/core/integration/tests/sdk/producer/mod.rs b/core/integration/tests/sdk/producer/mod.rs
index 464fdffa..455fec87 100644
--- a/core/integration/tests/sdk/producer/mod.rs
+++ b/core/integration/tests/sdk/producer/mod.rs
@@ -33,10 +33,7 @@ fn create_message_payload(offset: u64) -> Bytes {
async fn init_system(client: &IggyClient) {
// 1. Create the stream
- client
- .create_stream(STREAM_NAME)
- .await
- .unwrap();
+ client.create_stream(STREAM_NAME).await.unwrap();
// 2. Create the topic
client
diff --git a/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
index bd2a08f2..7d80f507 100644
--- a/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
@@ -17,8 +17,8 @@
*/
use crate::server::scenarios::{
- CONSUMER_GROUP_NAME, PARTITIONS_COUNT, STREAM_NAME,
- TOPIC_NAME, USERNAME_1, USERNAME_2, USERNAME_3, cleanup, create_client, join_consumer_group,
+ CONSUMER_GROUP_NAME, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME, USERNAME_1, USERNAME_2,
+ USERNAME_3, cleanup, create_client, join_consumer_group,
};
use iggy::clients::client::IggyClient;
use iggy::prelude::ClientInfoDetails;
@@ -42,10 +42,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
login_root(&system_client).await;
// 1. Create the stream
- let stream = system_client
- .create_stream(STREAM_NAME)
- .await
- .unwrap();
+ let stream = system_client.create_stream(STREAM_NAME).await.unwrap();
let stream_id = stream.id;
@@ -91,10 +88,12 @@ pub async fn run(client_factory: &dyn ClientFactory) {
join_consumer_group(&client1).await;
// 5. Get client1 info and validate that it contains the single consumer group
- let client1_info = get_me_and_validate_consumer_groups(&client1, stream_id, topic_id, consumer_group_id).await;
+ let client1_info =
+ get_me_and_validate_consumer_groups(&client1, stream_id, topic_id, consumer_group_id).await;
// 6. Validate that the consumer group has 1 member and this member has all partitions assigned
- let consumer_group = get_consumer_group_and_validate_members(&system_client, 1, consumer_group_id).await;
+ let consumer_group =
+ get_consumer_group_and_validate_members(&system_client, 1, consumer_group_id).await;
let member = &consumer_group.members[0];
assert_eq!(member.id, client1_info.client_id);
assert_eq!(member.partitions_count, PARTITIONS_COUNT);
@@ -107,7 +106,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
get_me_and_validate_consumer_groups(&client2, stream_id, topic_id, consumer_group_id).await;
// 9. Validate that the consumer group has 2 members and partitions are distributed between them
- let consumer_group = get_consumer_group_and_validate_members(&system_client, 2, consumer_group_id).await;
+ let consumer_group =
+ get_consumer_group_and_validate_members(&system_client, 2, consumer_group_id).await;
let member1 = &consumer_group.members[0];
let member2 = &consumer_group.members[1];
assert!(member1.partitions_count >= 1 && member1.partitions_count < PARTITIONS_COUNT);
@@ -124,7 +124,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
get_me_and_validate_consumer_groups(&client3, stream_id, topic_id, consumer_group_id).await;
// 12. Validate that the consumer group has 3 members and partitions are equally distributed between them
- let consumer_group = get_consumer_group_and_validate_members(&system_client, 3, consumer_group_id).await;
+ let consumer_group =
+ get_consumer_group_and_validate_members(&system_client, 3, consumer_group_id).await;
let member1 = &consumer_group.members[0];
let member2 = &consumer_group.members[1];
let member3 = &consumer_group.members[2];
@@ -140,10 +141,10 @@ pub async fn run(client_factory: &dyn ClientFactory) {
}
async fn get_me_and_validate_consumer_groups(
- client: &IggyClient,
- stream_id: u32,
- topic_id: u32,
- consumer_group_id: u32
+ client: &IggyClient,
+ stream_id: u32,
+ topic_id: u32,
+ consumer_group_id: u32,
) -> ClientInfoDetails {
let client_info = client.get_me().await.unwrap();
diff --git a/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
index f1e42c15..6d3b4cfe 100644
--- a/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
@@ -17,9 +17,8 @@
*/
use crate::server::scenarios::{
- CONSUMER_GROUP_NAME, MESSAGES_COUNT, PARTITIONS_COUNT,
- STREAM_NAME, TOPIC_NAME, cleanup, create_client, get_consumer_group,
- join_consumer_group,
+ CONSUMER_GROUP_NAME, MESSAGES_COUNT, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME, cleanup,
+ create_client, get_consumer_group, join_consumer_group,
};
use iggy::prelude::*;
use integration::test_server::{
@@ -50,10 +49,7 @@ async fn init_system(
create_users: bool,
) {
// 1. Create the stream
- system_client
- .create_stream(STREAM_NAME)
- .await
- .unwrap();
+ system_client.create_stream(STREAM_NAME).await.unwrap();
// 2. Create the topic
system_client
diff --git a/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
index ff3b0aff..58ef6c9c 100644
--- a/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
@@ -17,9 +17,8 @@
*/
use crate::server::scenarios::{
- CONSUMER_GROUP_NAME, MESSAGES_COUNT, PARTITIONS_COUNT,
- STREAM_NAME, TOPIC_NAME, cleanup, create_client, get_consumer_group,
- join_consumer_group,
+ CONSUMER_GROUP_NAME, MESSAGES_COUNT, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME, cleanup,
+ create_client, get_consumer_group, join_consumer_group,
};
use iggy::prelude::*;
use integration::test_server::{ClientFactory, assert_clean_system, login_root};
@@ -42,10 +41,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
async fn init_system(client: &IggyClient) {
// 1. Create the stream
- client
- .create_stream(STREAM_NAME)
- .await
- .unwrap();
+ client.create_stream(STREAM_NAME).await.unwrap();
// 2. Create the topic
client
diff --git a/core/integration/tests/server/scenarios/create_message_payload.rs b/core/integration/tests/server/scenarios/create_message_payload.rs
index 00f84e82..a35a18af 100644
--- a/core/integration/tests/server/scenarios/create_message_payload.rs
+++ b/core/integration/tests/server/scenarios/create_message_payload.rs
@@ -110,10 +110,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
async fn init_system(client: &IggyClient) {
// 1. Create the stream
- client
- .create_stream(STREAM_NAME)
- .await
- .unwrap();
+ client.create_stream(STREAM_NAME).await.unwrap();
// 2. Create the topic
client
diff --git a/core/integration/tests/server/scenarios/delete_segments_scenario.rs b/core/integration/tests/server/scenarios/delete_segments_scenario.rs
index 6fa4a38f..a35c5bf1 100644
--- a/core/integration/tests/server/scenarios/delete_segments_scenario.rs
+++ b/core/integration/tests/server/scenarios/delete_segments_scenario.rs
@@ -36,10 +36,7 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
.await
.unwrap();
- client
- .create_stream(STREAM_NAME)
- .await
- .unwrap();
+ client.create_stream(STREAM_NAME).await.unwrap();
client
.create_topic(
diff --git a/core/integration/tests/server/scenarios/encryption_scenario.rs b/core/integration/tests/server/scenarios/encryption_scenario.rs
index 1239583c..88c11585 100644
--- a/core/integration/tests/server/scenarios/encryption_scenario.rs
+++ b/core/integration/tests/server/scenarios/encryption_scenario.rs
@@ -77,10 +77,7 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
let topic_id = 1;
let partition_count = 1;
- client
- .create_stream("test-stream-api")
- .await
- .unwrap();
+ client.create_stream("test-stream-api").await.unwrap();
client
.create_topic(
&Identifier::numeric(stream_id).unwrap(),
@@ -89,7 +86,8 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
CompressionAlgorithm::default(),
None,
IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault)
+ MaxTopicSize::ServerDefault,
+ )
.await
.unwrap();
diff --git a/core/integration/tests/server/scenarios/message_headers_scenario.rs b/core/integration/tests/server/scenarios/message_headers_scenario.rs
index 21335618..89c9aebe 100644
--- a/core/integration/tests/server/scenarios/message_headers_scenario.rs
+++ b/core/integration/tests/server/scenarios/message_headers_scenario.rs
@@ -17,8 +17,7 @@
*/
use crate::server::scenarios::{
- MESSAGES_COUNT, PARTITION_ID, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME,
- cleanup, create_client,
+ MESSAGES_COUNT, PARTITION_ID, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME, cleanup, create_client,
};
use bytes::Bytes;
use iggy::prelude::*;
@@ -108,10 +107,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
async fn init_system(client: &IggyClient) {
// 1. Create the stream
- client
- .create_stream(STREAM_NAME)
- .await
- .unwrap();
+ client.create_stream(STREAM_NAME).await.unwrap();
// 2. Create the topic
client
diff --git a/core/integration/tests/server/scenarios/message_size_scenario.rs b/core/integration/tests/server/scenarios/message_size_scenario.rs
index 791cd2ce..a6430f59 100644
--- a/core/integration/tests/server/scenarios/message_size_scenario.rs
+++ b/core/integration/tests/server/scenarios/message_size_scenario.rs
@@ -109,10 +109,7 @@ async fn assert_message_count(client: &IggyClient, expected_count: u32) {
async fn init_system(client: &IggyClient) {
// 1. Create the stream
- client
- .create_stream(STREAM_NAME)
- .await
- .unwrap();
+ client.create_stream(STREAM_NAME).await.unwrap();
// 2. Create the topic
client
@@ -123,7 +120,7 @@ async fn init_system(client: &IggyClient) {
Default::default(),
None,
IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault
+ MaxTopicSize::ServerDefault,
)
.await
.unwrap();
diff --git a/core/integration/tests/server/scenarios/system_scenario.rs b/core/integration/tests/server/scenarios/system_scenario.rs
index 5e9967fc..ed27adbb 100644
--- a/core/integration/tests/server/scenarios/system_scenario.rs
+++ b/core/integration/tests/server/scenarios/system_scenario.rs
@@ -17,9 +17,8 @@
*/
use crate::server::scenarios::{
- CONSUMER_GROUP_NAME, CONSUMER_ID, CONSUMER_KIND, MESSAGES_COUNT,
- PARTITION_ID, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME,
- get_consumer_group, leave_consumer_group,
+ CONSUMER_GROUP_NAME, CONSUMER_ID, CONSUMER_KIND, MESSAGES_COUNT, PARTITION_ID,
+ PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME, get_consumer_group, leave_consumer_group,
};
use bytes::Bytes;
use iggy::prelude::*;
@@ -49,10 +48,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
assert!(streams.is_empty());
// 3. Create the stream
- let stream = client
- .create_stream(STREAM_NAME)
- .await
- .unwrap();
+ let stream = client.create_stream(STREAM_NAME).await.unwrap();
let stream_id = stream.id;
assert_eq!(stream.name, STREAM_NAME);
@@ -92,9 +88,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
assert_eq!(stream.name, STREAM_NAME);
// 7. Try to create the stream with the same name and validate that it fails
- let create_stream_result = client
- .create_stream(STREAM_NAME)
- .await;
+ let create_stream_result = client.create_stream(STREAM_NAME).await;
assert!(create_stream_result.is_err());
// 8. Try to create the stream with the different name and validate that it succeeds
diff --git a/core/integration/tests/server/scenarios/tcp_tls_scenario.rs b/core/integration/tests/server/scenarios/tcp_tls_scenario.rs
index 947fcf08..bac2b185 100644
--- a/core/integration/tests/server/scenarios/tcp_tls_scenario.rs
+++ b/core/integration/tests/server/scenarios/tcp_tls_scenario.rs
@@ -24,10 +24,7 @@ pub async fn run(client: &IggyClient) {
login_root(client).await;
let stream_name = "test-tls-stream";
- client
- .create_stream(stream_name)
- .await
- .unwrap();
+ client.create_stream(stream_name).await.unwrap();
let stream = client
.get_stream(&Identifier::named(stream_name).unwrap())
diff --git a/core/integration/tests/state/system.rs b/core/integration/tests/state/system.rs
index 63079d57..dce78aea 100644
--- a/core/integration/tests/state/system.rs
+++ b/core/integration/tests/state/system.rs
@@ -240,18 +240,12 @@ async fn should_be_initialized_based_on_state_entries() {
);
assert_eq!(system.streams.len(), 1);
- let mut stream = system
- .streams
- .remove(&stream1_id)
- .unwrap();
+ let mut stream = system.streams.remove(&stream1_id).unwrap();
assert_eq!(stream.id, stream1_id);
assert_eq!(stream.name, create_stream1_clone.name);
assert_eq!(stream.topics.len(), 1);
- let mut topic = stream
- .topics
- .remove(&topic1_id)
- .unwrap();
+ let mut topic = stream.topics.remove(&topic1_id).unwrap();
assert_eq!(topic.id, topic1_id);
assert_eq!(topic.name, create_topic1_clone.name);
assert_eq!(topic.partitions.len(), 3);
@@ -259,9 +253,6 @@ async fn should_be_initialized_based_on_state_entries() {
assert_eq!(topic.consumer_groups.len(), 1);
let consumer_group = topic.consumer_groups.remove(&group_id).unwrap();
- assert_eq!(
- consumer_group.id,
- group_id
- );
+ assert_eq!(consumer_group.id, group_id);
assert_eq!(consumer_group.name, create_consumer_group_clone.name);
}
diff --git a/core/sdk/src/client_wrappers/binary_stream_client.rs b/core/sdk/src/client_wrappers/binary_stream_client.rs
index e1ad849a..78291668 100644
--- a/core/sdk/src/client_wrappers/binary_stream_client.rs
+++ b/core/sdk/src/client_wrappers/binary_stream_client.rs
@@ -41,10 +41,7 @@ impl StreamClient for ClientWrapper {
}
}
- async fn create_stream(
- &self,
- name: &str,
- ) -> Result<StreamDetails, IggyError> {
+ async fn create_stream(&self, name: &str) -> Result<StreamDetails, IggyError> {
match self {
ClientWrapper::Iggy(client) => client.create_stream(name).await,
ClientWrapper::Http(client) => client.create_stream(name).await,
diff --git a/core/sdk/src/clients/binary_streams.rs b/core/sdk/src/clients/binary_streams.rs
index 74cf30ac..b64e8c6c 100644
--- a/core/sdk/src/clients/binary_streams.rs
+++ b/core/sdk/src/clients/binary_streams.rs
@@ -32,15 +32,8 @@ impl StreamClient for IggyClient {
self.client.read().await.get_streams().await
}
- async fn create_stream(
- &self,
- name: &str,
- ) -> Result<StreamDetails, IggyError> {
- self.client
- .read()
- .await
- .create_stream(name)
- .await
+ async fn create_stream(&self, name: &str) -> Result<StreamDetails, IggyError> {
+ self.client.read().await.create_stream(name).await
}
async fn update_stream(&self, stream_id: &Identifier, name: &str) -> Result<(), IggyError> {
diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs
index 66975eae..273f2cb8 100644
--- a/core/sdk/src/clients/consumer.rs
+++ b/core/sdk/src/clients/consumer.rs
@@ -891,7 +891,6 @@ impl ReceivedMessage {
}
}
-
impl Stream for IggyConsumer {
type Item = Result<ReceivedMessage, IggyError>;
@@ -1028,4 +1027,4 @@ impl Stream for IggyConsumer {
Poll::Pending
}
-}
\ No newline at end of file
+}
diff --git a/core/sdk/src/http/binary_streams.rs b/core/sdk/src/http/binary_streams.rs
index fe6199de..509cbd59 100644
--- a/core/sdk/src/http/binary_streams.rs
+++ b/core/sdk/src/http/binary_streams.rs
@@ -56,10 +56,7 @@ impl StreamClient for HttpClient {
Ok(streams)
}
- async fn create_stream(
- &self,
- name: &str,
- ) -> Result<StreamDetails, IggyError> {
+ async fn create_stream(&self, name: &str) -> Result<StreamDetails, IggyError> {
let response = self
.post(
PATH,
diff --git a/core/server/src/binary/handlers/streams/create_stream_handler.rs b/core/server/src/binary/handlers/streams/create_stream_handler.rs
index 21d24ff6..0b1c1ec0 100644
--- a/core/server/src/binary/handlers/streams/create_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/create_stream_handler.rs
@@ -48,9 +48,7 @@ impl ServerCommandHandler for CreateStream {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
- let stream = shard
- .create_stream2(session, self.name.clone())
- .await?;
+ let stream = shard.create_stream2(session, self.name.clone()).await?;
let created_stream_id = stream.id();
shard_info!(
shard.id,
diff --git a/core/server/src/binary/mapper.rs b/core/server/src/binary/mapper.rs
index 340c9f90..fa933aa4 100644
--- a/core/server/src/binary/mapper.rs
+++ b/core/server/src/binary/mapper.rs
@@ -20,7 +20,7 @@ use std::sync::{Arc, atomic::AtomicU64};
use crate::slab::Keyed;
use crate::slab::traits_ext::{EntityComponentSystem, IntoComponents};
-use crate::streaming::clients::client_manager::{Client};
+use crate::streaming::clients::client_manager::Client;
use crate::streaming::partitions::partition2::PartitionRoot;
use crate::streaming::personal_access_tokens::personal_access_token::PersonalAccessToken;
use crate::streaming::stats::stats::{PartitionStats, StreamStats, TopicStats};
@@ -30,9 +30,7 @@ use crate::streaming::topics::topic2::{self, TopicRoot};
use crate::streaming::users::user::User;
use arcshift::SharedGetGuard;
use bytes::{BufMut, Bytes, BytesMut};
-use iggy_common::{
- BytesSerializable, ConsumerOffsetInfo, Stats, TransportProtocol, UserId,
-};
+use iggy_common::{BytesSerializable, ConsumerOffsetInfo, Stats, TransportProtocol, UserId};
use slab::Slab;
pub fn map_stats(stats: &Stats) -> Bytes {
diff --git a/core/server/src/channels/commands/clean_personal_access_tokens.rs b/core/server/src/channels/commands/clean_personal_access_tokens.rs
index 12153356..d846aa27 100644
--- a/core/server/src/channels/commands/clean_personal_access_tokens.rs
+++ b/core/server/src/channels/commands/clean_personal_access_tokens.rs
@@ -72,4 +72,4 @@ pub async fn clear_personal_access_tokens(shard: Rc<IggyShard>) -> Result<(), Ig
deleted_tokens_count
);
}
-}
\ No newline at end of file
+}
diff --git a/core/server/src/channels/commands/print_sysinfo.rs b/core/server/src/channels/commands/print_sysinfo.rs
index 05f68372..0c3992d2 100644
--- a/core/server/src/channels/commands/print_sysinfo.rs
+++ b/core/server/src/channels/commands/print_sysinfo.rs
@@ -67,4 +67,4 @@ pub async fn print_sys_info(shard: Rc<IggyShard>) -> Result<(), IggyError> {
memory_pool().log_stats();
}
-}
\ No newline at end of file
+}
diff --git a/core/server/src/channels/commands/save_messages.rs b/core/server/src/channels/commands/save_messages.rs
index 51731042..84330563 100644
--- a/core/server/src/channels/commands/save_messages.rs
+++ b/core/server/src/channels/commands/save_messages.rs
@@ -74,7 +74,11 @@ pub async fn save_messages(shard: Rc<IggyShard>) -> Result<(), IggyError> {
}
if total_saved_messages > 0 {
- shard_info!(shard.id, "Saved {} buffered messages on disk.", total_saved_messages);
+ shard_info!(
+ shard.id,
+ "Saved {} buffered messages on disk.",
+ total_saved_messages
+ );
}
trace!("Finished saving buffered messages.");
diff --git a/core/server/src/http/system.rs b/core/server/src/http/system.rs
index d4e2b0f6..3741ec6a 100644
--- a/core/server/src/http/system.rs
+++ b/core/server/src/http/system.rs
@@ -79,7 +79,8 @@ async fn get_cluster_metadata(
) -> Result<Json<ClusterMetadata>, CustomError> {
let shard = state.shard.shard();
let session = Session::stateless(identity.user_id, identity.ip_address);
- let cluster_metadata = shard.get_cluster_metadata(&session)
+ let cluster_metadata = shard
+ .get_cluster_metadata(&session)
.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - failed to get cluster metadata, user ID: {}",
diff --git a/core/server/src/quic/listener.rs b/core/server/src/quic/listener.rs
index ba244437..5fce5c84 100644
--- a/core/server/src/quic/listener.rs
+++ b/core/server/src/quic/listener.rs
@@ -75,8 +75,7 @@ async fn handle_connection(
) -> Result<(), ConnectionError> {
let address = connection.remote_address();
info!("Client has connected: {address}");
- let session = shard
- .add_client(&address, TransportProtocol::Quic);
+ let session = shard.add_client(&address, TransportProtocol::Quic);
let session = shard.add_client(&address, TransportProtocol::Quic);
let client_id = session.client_id;
diff --git a/core/server/src/shard/system/clients.rs b/core/server/src/shard/system/clients.rs
index fe4a1cde..bf8ebe17 100644
--- a/core/server/src/shard/system/clients.rs
+++ b/core/server/src/shard/system/clients.rs
@@ -18,12 +18,12 @@
use super::COMPONENT;
use crate::shard::IggyShard;
-use crate::streaming::clients::client_manager::{Client};
+use crate::streaming::clients::client_manager::Client;
use crate::streaming::session::Session;
use error_set::ErrContext;
-use iggy_common::{Identifier, TransportProtocol};
use iggy_common::IggyError;
use iggy_common::locking::IggyRwLockFn;
+use iggy_common::{Identifier, TransportProtocol};
use std::net::SocketAddr;
use std::rc::Rc;
use tracing::{error, info};
diff --git a/core/server/src/shard/system/cluster.rs b/core/server/src/shard/system/cluster.rs
index f2bca11b..1619bb8a 100644
--- a/core/server/src/shard/system/cluster.rs
+++ b/core/server/src/shard/system/cluster.rs
@@ -18,7 +18,9 @@
use crate::shard::IggyShard;
use crate::streaming::session::Session;
-use iggy_common::{ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, IggyError, TransportProtocol};
+use iggy_common::{
+ ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, IggyError, TransportProtocol,
+};
use std::str::FromStr;
use tracing::trace;
@@ -71,4 +73,4 @@ impl IggyShard {
nodes,
})
}
-}
\ No newline at end of file
+}
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index 0cf4e4ca..d7e725b1 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -9,7 +9,8 @@ use crate::streaming::{
},
};
use iggy_common::{
- CompressionAlgorithm, Identifier, IggyExpiry, MaxTopicSize, Permissions, TransportProtocol, UserStatus
+ CompressionAlgorithm, Identifier, IggyExpiry, MaxTopicSize, Permissions, TransportProtocol,
+ UserStatus,
};
use std::net::SocketAddr;
diff --git a/core/server/src/streaming/clients/client_manager.rs b/core/server/src/streaming/clients/client_manager.rs
index 74cc42c3..8e41b69c 100644
--- a/core/server/src/streaming/clients/client_manager.rs
+++ b/core/server/src/streaming/clients/client_manager.rs
@@ -49,7 +49,11 @@ pub struct ConsumerGroup {
}
impl ClientManager {
- pub fn add_client(&mut self, address: &SocketAddr, transport: TransportProtocol) -> Rc<Session> {
+ pub fn add_client(
+ &mut self,
+ address: &SocketAddr,
+ transport: TransportProtocol,
+ ) -> Rc<Session> {
let client_id = hash::calculate_32(address.to_string().as_bytes());
let session = Rc::new(Session::from_client_id(client_id, *address));
let client = Client {
diff --git a/core/server/src/streaming/cluster/mod.rs b/core/server/src/streaming/cluster/mod.rs
deleted file mode 100644
index 036ebe1c..00000000
--- a/core/server/src/streaming/cluster/mod.rs
+++ /dev/null
@@ -1,71 +0,0 @@
-/* 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 crate::streaming::session::Session;
-use crate::streaming::systems::system::System;
-use iggy_common::IggyError;
-use iggy_common::metadata::ClusterMetadata;
-use iggy_common::node::ClusterNode;
-use iggy_common::role::ClusterNodeRole;
-use iggy_common::status::ClusterNodeStatus;
-use tracing::trace;
-
-impl System {
- pub fn get_cluster_metadata(&self, session: &Session) -> Result<ClusterMetadata, IggyError> {
- trace!("Getting cluster metadata for session: {session}");
-
- // TODO(hubcio): Clustering is not yet implemented
- // The leader/follower as well as node status are currently placeholder implementations.
-
- let nodes: Vec<ClusterNode> = self
- .cluster_config
- .nodes
- .iter()
- .map(|node_config| {
- let role = if node_config.id == 1 {
- ClusterNodeRole::Leader
- } else {
- ClusterNodeRole::Follower
- };
-
- let status = if self.cluster_config.enabled {
- ClusterNodeStatus::Healthy
- } else {
- ClusterNodeStatus::Stopping
- };
-
- ClusterNode {
- id: node_config.id,
- name: node_config.name.clone(),
- address: node_config.address.clone(),
- role,
- status,
- }
- })
- .collect();
-
- Ok(ClusterMetadata {
- name: self.cluster_config.name.clone(),
- id: self.cluster_config.id,
- transport: self.cluster_config.transport.clone(),
- nodes,
- })
- }
-}
- */
diff --git a/core/server/src/streaming/mod.rs b/core/server/src/streaming/mod.rs
index ee36cc4a..fe1eb061 100644
--- a/core/server/src/streaming/mod.rs
+++ b/core/server/src/streaming/mod.rs
@@ -17,7 +17,6 @@
*/
pub mod clients;
-pub mod cluster;
pub mod deduplication;
pub mod diagnostics;
pub mod partitions;
diff --git a/core/tools/src/data-seeder/seeder.rs b/core/tools/src/data-seeder/seeder.rs
index ec536317..bbd176fb 100644
--- a/core/tools/src/data-seeder/seeder.rs
+++ b/core/tools/src/data-seeder/seeder.rs
@@ -34,16 +34,16 @@ pub async fn seed(client: &IggyClient) -> Result<(), IggyError> {
async fn create_streams(client: &IggyClient) -> Result<Vec<(String, u32)>, IggyError> {
let mut streams = Vec::new();
-
+
let prod_stream = client.create_stream(PROD_STREAM_NAME).await?;
streams.push((PROD_STREAM_NAME.to_string(), prod_stream.id));
-
+
let test_stream = client.create_stream(TEST_STREAM_NAME).await?;
streams.push((TEST_STREAM_NAME.to_string(), test_stream.id));
-
+
let dev_stream = client.create_stream(DEV_STREAM_NAME).await?;
streams.push((DEV_STREAM_NAME.to_string(), dev_stream.id));
-
+
Ok(streams)
}
diff --git a/examples/rust/src/getting-started/producer/main.rs b/examples/rust/src/getting-started/producer/main.rs
index 6114bc74..2e8ec22d 100644
--- a/examples/rust/src/getting-started/producer/main.rs
+++ b/examples/rust/src/getting-started/producer/main.rs
@@ -62,7 +62,8 @@ async fn init_system(client: &IggyClient) -> (u32, u32) {
}
Err(_) => {
warn!("Stream already exists and will not be created again.");
- client.get_stream(&Identifier::named(STREAM_NAME).unwrap())
+ client
+ .get_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap()
.expect("Failed to get stream")
@@ -87,20 +88,25 @@ async fn init_system(client: &IggyClient) -> (u32, u32) {
}
Err(_) => {
warn!("Topic already exists and will not be created again.");
- client.get_topic(
- &Identifier::named(STREAM_NAME).unwrap(),
- &Identifier::named(TOPIC_NAME).unwrap(),
- )
- .await
- .unwrap()
- .expect("Failed to get topic")
+ client
+ .get_topic(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ )
+ .await
+ .unwrap()
+ .expect("Failed to get topic")
}
};
(stream.id, topic.id)
}
-async fn produce_messages(client: &dyn Client, stream_id: u32, topic_id: u32) -> Result<(), Box<dyn Error>> {
+async fn produce_messages(
+ client: &dyn Client,
+ stream_id: u32,
+ topic_id: u32,
+) -> Result<(), Box<dyn Error>> {
let duration = IggyDuration::from_str("500ms")?;
let mut interval = tokio::time::interval(duration.get_duration());
info!(