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/08/14 14:00:44 UTC
(iggy) branch remove_old_api created (now c4933d10)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch remove_old_api
in repository https://gitbox.apache.org/repos/asf/iggy.git
at c4933d10 part1
This branch includes the following new commits:
new c4933d10 part1
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: part1
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch remove_old_api
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c4933d10233c2454ca74f800cfced17858e8da57
Author: numminex <g....@gmail.com>
AuthorDate: Wed Aug 13 22:17:45 2025 +0200
part1
---
.../handlers/streams/create_stream_handler.rs | 59 +--
.../binary/handlers/streams/get_stream_handler.rs | 32 +-
.../binary/handlers/streams/get_streams_handler.rs | 21 +-
.../handlers/streams/update_stream_handler.rs | 2 +-
.../binary/handlers/topics/create_topic_handler.rs | 109 +---
.../binary/handlers/topics/delete_topic_handler.rs | 68 +--
.../binary/handlers/topics/get_topic_handler.rs | 34 +-
.../binary/handlers/topics/get_topics_handler.rs | 39 +-
.../binary/handlers/topics/update_topic_handler.rs | 56 +-
core/server/src/binary/mapper.rs | 127 +++--
core/server/src/shard/mod.rs | 36 +-
core/server/src/shard/system/partitions.rs | 4 +-
core/server/src/shard/system/streams.rs | 477 +----------------
core/server/src/shard/system/topics.rs | 583 ++-------------------
core/server/src/shard/system/utils.rs | 9 +-
core/server/src/slab/helpers.rs | 50 ++
core/server/src/slab/mod.rs | 21 +-
core/server/src/slab/partitions.rs | 10 +-
core/server/src/slab/streams.rs | 179 +++----
core/server/src/slab/topics.rs | 95 ++--
core/server/src/slab/traits_ext.rs | 55 +-
core/server/src/streaming/partitions/partition2.rs | 4 +
core/server/src/streaming/stats/stats.rs | 34 +-
core/server/src/streaming/streams/helpers.rs | 18 +
core/server/src/streaming/streams/mod.rs | 1 +
core/server/src/streaming/streams/stream2.rs | 4 +
core/server/src/streaming/streams/topics.rs | 205 --------
core/server/src/streaming/topics/helpers.rs | 58 ++
core/server/src/streaming/topics/mod.rs | 1 +
core/server/src/streaming/topics/storage.rs | 182 +------
core/server/src/streaming/topics/topic.rs | 30 +-
core/server/src/streaming/topics/topic2.rs | 47 +-
32 files changed, 678 insertions(+), 1972 deletions(-)
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 95d960f5..4da9950e 100644
--- a/core/server/src/binary/handlers/streams/create_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/create_stream_handler.rs
@@ -23,17 +23,15 @@ use crate::binary::{handlers::streams::COMPONENT, sender::SenderKind};
use crate::shard::IggyShard;
use crate::shard::transmission::event::ShardEvent;
use crate::shard_info;
-use crate::slab::traits_ext::EntityMarker;
+use crate::slab::traits_ext::{EntityComponentSystem, EntityMarker};
use crate::state::command::EntryCommand;
use crate::state::models::CreateStreamWithId;
use crate::streaming::session::Session;
-use crate::streaming::stats::stats::StreamStats;
use anyhow::Result;
use error_set::ErrContext;
+use iggy_common::IggyError;
use iggy_common::create_stream::CreateStream;
-use iggy_common::{Identifier, IggyError};
use std::rc::Rc;
-use std::sync::Arc;
use tracing::{debug, instrument};
impl ServerCommandHandler for CreateStream {
@@ -51,55 +49,34 @@ impl ServerCommandHandler for CreateStream {
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
let stream_id = self.stream_id;
- let name = self.name.clone();
-
let stream = shard
.create_stream2(session, stream_id, self.name.clone())
.await?;
+ let created_stream_id = stream.id();
shard_info!(
shard.id,
- "Created stream with new API, Stream ID: {}, name: '{}'.",
- stream.id(),
- name
+ "Created stream with ID: {}, name: '{}'.",
+ created_stream_id,
+ stream.root().name()
);
let event = ShardEvent::CreatedStream2 {
- id: stream.id(),
+ id: created_stream_id,
stream,
};
- let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
-
- let created_stream_id = shard
- .create_stream(session, stream_id, &name)
- .await
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to create stream with id: {stream_id:?}, session: {session}"
- )
- })?;
- shard_info!(
- shard.id,
- "Created stream with ID: {}, name: '{}'.",
- created_stream_id,
- name
- );
- let event = ShardEvent::CreatedStream { stream_id, name };
- // Broadcast the event to all shards.
- let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
-
- let stream = shard.find_stream(session, &created_stream_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to find created stream with id: {created_stream_id:?}, session: {session}"
- )
- })?;
- let response = mapper::map_stream(&stream);
+ let _responses = shard.broadcast_event_to_all_shards(event).await;
+ let response = shard
+ .streams2
+ .with_by_id(created_stream_id, |(root, stats)| {
+ mapper::map_stream(&root, &stats)
+ });
shard
.state
- .apply(session.get_user_id(), &EntryCommand::CreateStream(CreateStreamWithId {
- stream_id: stream.stream_id,
- command: self
- })) .await
+ .apply(session.get_user_id(), &EntryCommand::CreateStream(CreateStreamWithId {
+ stream_id: created_stream_id as u32,
+ command: self
+ }))
+ .await
.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - failed to apply create stream for id: {stream_id:?}, session: {session}"
diff --git a/core/server/src/binary/handlers/streams/get_stream_handler.rs b/core/server/src/binary/handlers/streams/get_stream_handler.rs
index fef6458c..15a9500f 100644
--- a/core/server/src/binary/handlers/streams/get_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/get_stream_handler.rs
@@ -21,8 +21,10 @@ use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::binary::sender::SenderKind;
use crate::shard::IggyShard;
+use crate::slab::traits_ext::EntityComponentSystem;
use crate::streaming::session::Session;
use anyhow::Result;
+use error_set::ErrContext;
use iggy_common::IggyError;
use iggy_common::get_stream::GetStream;
use std::rc::Rc;
@@ -41,17 +43,25 @@ impl ServerCommandHandler for GetStream {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
- let Ok(stream) = shard.try_find_stream(session, &self.stream_id) else {
- sender.send_empty_ok_response().await?;
- return Ok(());
- };
-
- let Some(stream) = stream else {
- sender.send_empty_ok_response().await?;
- return Ok(());
- };
-
- let response = mapper::map_stream(&stream);
+ shard.ensure_authenticated(session)?;
+ shard.ensure_stream_exists(&self.stream_id)?;
+ let stream_id = shard
+ .streams2
+ .with_root_by_id(&self.stream_id, |root| root.id());
+ shard
+ .permissioner
+ .borrow()
+ .get_stream(session.get_user_id(), stream_id as u32)
+ .with_error_context(|_| {
+ format!(
+ "permission denied to get stream with ID: {} for user with ID: {}",
+ self.stream_id,
+ session.get_user_id(),
+ )
+ });
+ let response = shard
+ .streams2
+ .with_by_id(stream_id, |(root, stats)| mapper::map_stream(&root, &stats));
sender.send_ok_response(&response).await?;
Ok(())
}
diff --git a/core/server/src/binary/handlers/streams/get_streams_handler.rs b/core/server/src/binary/handlers/streams/get_streams_handler.rs
index 0f83a9cb..9010a383 100644
--- a/core/server/src/binary/handlers/streams/get_streams_handler.rs
+++ b/core/server/src/binary/handlers/streams/get_streams_handler.rs
@@ -22,6 +22,7 @@ use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::binary::sender::SenderKind;
use crate::shard::IggyShard;
+use crate::slab::traits_ext::{EntityComponentSystem, IntoComponents};
use crate::streaming::session::Session;
use anyhow::Result;
use error_set::ErrContext;
@@ -43,10 +44,22 @@ impl ServerCommandHandler for GetStreams {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
- let streams = shard.find_streams(session).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to find streams for session: {session}")
- })?;
- let response = mapper::map_streams(streams);
+ shard.ensure_authenticated(session)?;
+ shard
+ .permissioner
+ .borrow()
+ .get_streams(session.get_user_id())
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - permission denied to get streams for user {}",
+ session.get_user_id()
+ )
+ })?;
+
+ let response = shard.streams2.with(|stream_ref| {
+ let (roots, stats) = stream_ref.into_components();
+ mapper::map_streams(&roots, &stats)
+ });
sender.send_ok_response(&response).await?;
Ok(())
}
diff --git a/core/server/src/binary/handlers/streams/update_stream_handler.rs b/core/server/src/binary/handlers/streams/update_stream_handler.rs
index bb4d10b1..f8ca41ab 100644
--- a/core/server/src/binary/handlers/streams/update_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/update_stream_handler.rs
@@ -47,7 +47,7 @@ impl ServerCommandHandler for UpdateStream {
let stream_id = self.stream_id.clone();
shard
- .update_stream2(session, &self.stream_id, &self.name)
+ .update_stream2(session, &self.stream_id, self.name.clone())
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to update stream2 with id: {stream_id}, session: {session}")
})?;
diff --git a/core/server/src/binary/handlers/topics/create_topic_handler.rs b/core/server/src/binary/handlers/topics/create_topic_handler.rs
index 3f913685..223ba118 100644
--- a/core/server/src/binary/handlers/topics/create_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/create_topic_handler.rs
@@ -20,24 +20,17 @@ use crate::binary::command::{BinaryServerCommand, ServerCommand, ServerCommandHa
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::binary::{handlers::topics::COMPONENT, sender::SenderKind};
-use crate::shard::namespace::IggyNamespace;
+use crate::shard::IggyShard;
use crate::shard::transmission::event::ShardEvent;
-use crate::shard::{IggyShard, ShardInfo};
-use crate::shard_info;
use crate::slab::traits_ext::EntityMarker;
use crate::state::command::EntryCommand;
use crate::state::models::CreateTopicWithId;
-use crate::state::system::TopicState;
use crate::streaming::session::Session;
-use crate::streaming::stats::stats::TopicStats;
-use crate::streaming::topics::topic::Topic;
use anyhow::Result;
use error_set::ErrContext;
use iggy_common::create_topic::CreateTopic;
-use iggy_common::locking::IggyRwLockFn;
use iggy_common::{Identifier, IggyError};
use std::rc::Rc;
-use std::sync::Arc;
use tracing::{debug, instrument};
impl ServerCommandHandler for CreateTopic {
@@ -54,13 +47,10 @@ impl ServerCommandHandler for CreateTopic {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
- let stream_id = self.stream_id.clone();
- let maybe_topic_id = self.topic_id;
let topic = shard
.create_topic2(
session,
- &stream_id,
- maybe_topic_id,
+ &self.stream_id,
self.name.clone(),
self.message_expiry,
self.compression_algorithm,
@@ -68,6 +58,12 @@ impl ServerCommandHandler for CreateTopic {
self.replication_factor,
)
.await?;
+ self.message_expiry = topic.root().message_expiry();
+ self.max_topic_size = topic.root().max_topic_size();
+
+ let stream_id = shard
+ .streams2
+ .with_root_by_id(&self.stream_id, |root| root.id());
let topic_id = topic.id();
// Send events for topic creation.
let event = ShardEvent::CreatedTopic2 {
@@ -79,96 +75,21 @@ impl ServerCommandHandler for CreateTopic {
shard
.create_partitions2(
session,
- &stream_id,
+ &self.stream_id,
&Identifier::numeric(topic_id as u32).unwrap(),
self.partitions_count,
)
.await?;
- let (topic_id, partition_ids) = shard
- .create_topic(
- session,
- &stream_id,
- maybe_topic_id,
- &self.name,
- self.partitions_count,
- self.message_expiry,
- self.compression_algorithm,
- self.max_topic_size,
- self.replication_factor,
- )
- .await
- .with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to create topic for stream_id: {stream_id}, topic_id: {maybe_topic_id:?}"
- ))?;
-
- let message_expiry = Topic::get_message_expiry(self.message_expiry, &shard.config.system);
- shard_info!(
- shard.id,
- "Received message expiry: {}, set expiry: {}",
- self.message_expiry,
- message_expiry
+ // TODO: Create shard_table records for partitions.
+ let response = shard.streams2.with_topic_by_id(
+ &self.stream_id,
+ &Identifier::numeric(topic_id as u32).unwrap(),
+ |(root, stats)| mapper::map_topic(&root, &stats),
);
- shard_info!(shard.id, "Created topic with ID: {}", topic_id);
-
- // `create_topic` always returns numeric identifier.
- let numeric_topic_id = topic_id.get_u32_value().unwrap();
- let numeric_stream_id = shard.get_stream(&stream_id).map(|s| s.stream_id).unwrap();
- let event = ShardEvent::CreatedTopic {
- stream_id: stream_id.clone(),
- topic_id: topic_id.clone(),
- name: self.name.clone(),
- partitions_count: self.partitions_count,
- message_expiry: self.message_expiry,
- compression_algorithm: self.compression_algorithm,
- max_topic_size: self.max_topic_size,
- replication_factor: self.replication_factor,
- };
- let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
- let stream = shard.get_stream(&stream_id).with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to get stream for stream_id: {stream_id}"
- )
- })?;
- let topic = stream.get_topic(&topic_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id} in stream with ID: {stream_id}"
- )
- })?;
-
- let records = shard
- .create_shard_table_records(&partition_ids, numeric_stream_id, numeric_topic_id)
- .collect::<Vec<_>>();
- // Open partition and segments for that particular shard.
- for (ns, shard_info) in records.iter() {
- if shard_info.id() == shard.id {
- let partition_id = ns.partition_id;
- let partition = topic.get_partition(partition_id)?;
- let mut partition = partition.write().await;
- partition.open().await.with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to open partition with ID: {partition_id} in topic with ID: {topic_id} for stream with ID: {stream_id}"
- )
- })?;
- }
- }
- shard.insert_shard_table_records(records);
- // Broadcast the event to all shards.
- let event = ShardEvent::CreatedShardTableRecords {
- stream_id: numeric_stream_id,
- topic_id: numeric_topic_id,
- partition_ids,
- };
- let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
-
- self.message_expiry = topic.message_expiry;
- self.max_topic_size = topic.max_topic_size;
- let topic_id = topic.topic_id;
- let response = mapper::map_topic(topic).await;
-
shard
.state
.apply(session.get_user_id(), &EntryCommand::CreateTopic(CreateTopicWithId {
- topic_id,
+ topic_id: topic_id as u32,
command: self
}))
.await
diff --git a/core/server/src/binary/handlers/topics/delete_topic_handler.rs b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
index 70fccd8c..bc7f4560 100644
--- a/core/server/src/binary/handlers/topics/delete_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
@@ -19,23 +19,16 @@
use crate::binary::command::{BinaryServerCommand, ServerCommand, ServerCommandHandler};
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::{handlers::topics::COMPONENT, sender::SenderKind};
-use crate::io::fs_utils::remove_dir_all;
use crate::shard::IggyShard;
-use crate::shard::namespace::IggyNamespace;
use crate::shard::transmission::event::ShardEvent;
use crate::shard_info;
-use crate::slab::traits_ext::EntityMarker;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use anyhow::Result;
-use compio::driver::op::OpenFile;
-use compio::fs::File;
use error_set::ErrContext;
use iggy_common::IggyError;
use iggy_common::delete_topic::DeleteTopic;
-use iggy_common::locking::IggyRwLockFn;
use std::rc::Rc;
-use std::time::Duration;
use tracing::{debug, instrument};
impl ServerCommandHandler for DeleteTopic {
@@ -57,65 +50,36 @@ impl ServerCommandHandler for DeleteTopic {
// we would end up in a state where the topic is deleted from disk, but during state recreation it would be recreated,
// without persisted messages in the partitions.
debug!("session: {session}, command: {self}");
- let topic2 = shard
+ let topic = shard
.delete_topic2(session, &self.stream_id, &self.topic_id)
.await?;
+ let stream_id = shard
+ .streams2
+ .with_root_by_id(&self.stream_id, |root| root.id());
+ let topic_id = topic.root().id();
+ shard_info!(
+ shard.id,
+ "Deleted topic with name: {}, ID: {} in stream with ID: {}",
+ topic.root().name(),
+ topic_id,
+ stream_id
+ );
let event = ShardEvent::DeletedTopic2 {
- id: topic2.id(),
+ id: topic_id,
stream_id: self.stream_id.clone(),
topic_id: self.topic_id.clone(),
};
let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
// Drop the topic to force readers/writers to be dropped.
- drop(topic2);
- // Remove all the files and directories.
-
- let topic = shard
- .delete_topic(session, &self.stream_id, &self.topic_id)
- .await
- .with_error_context(|error| format!(
- "{COMPONENT} (error: {error}) - failed to delete topic with ID: {} in stream with ID: {}, session: {session}",
- &self.topic_id, &self.stream_id
- ))?;
- shard_info!(
- shard.id,
- "Deleted topic with name: {}, ID: {} in stream with ID: {}",
- topic.name,
- topic.topic_id,
- topic.stream_id
- );
- let numeric_topic_id = topic.topic_id;
- let numeric_stream_id = topic.stream_id;
- let partitions = topic.get_partitions();
- let mut namespaces = Vec::new();
- for partition in partitions {
- let partition_id = partition.read().await.partition_id;
- let namespace = IggyNamespace::new(numeric_stream_id, topic.topic_id, partition_id);
- namespaces.push(namespace);
- }
- let records = shard.remove_shard_table_records(&namespaces);
- for (ns, shard_info) in records {
- if shard_info.id() == shard.id {
- let partition = topic.get_partition(ns.partition_id).unwrap();
- let mut partition = partition.write().await;
- let partition_id = partition.partition_id;
- partition.delete().await.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to delete partition in stream: {self}, partition: {partition_id}")
- })?;
- }
- }
- let event = ShardEvent::DeletedShardTableRecords { namespaces };
- let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
- topic.delete().await.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to delete topic in stream: {self}")
- })?;
+ drop(topic);
+ // TODO: Remove all the files and directories.
shard
.state
.apply(session.get_user_id(), &EntryCommand::DeleteTopic(self))
.await
.with_error_context(|error| format!(
- "{COMPONENT} (error: {error}) - failed to apply delete topic with ID: {numeric_topic_id} in stream with ID: {numeric_stream_id}, session: {session}",
+ "{COMPONENT} (error: {error}) - failed to apply delete topic with ID: {topic_id} in stream with ID: {stream_id}, session: {session}",
))?;
sender.send_empty_ok_response().await?;
Ok(())
diff --git a/core/server/src/binary/handlers/topics/get_topic_handler.rs b/core/server/src/binary/handlers/topics/get_topic_handler.rs
index 597c13bf..7c815a43 100644
--- a/core/server/src/binary/handlers/topics/get_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/get_topic_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use anyhow::Result;
-use error_set::ErrContext;
use iggy_common::IggyError;
use iggy_common::get_topic::GetTopic;
use std::rc::Rc;
@@ -42,26 +41,25 @@ impl ServerCommandHandler for GetTopic {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
- let stream = shard
- .find_stream(session, &self.stream_id)
- .with_error_context(|error| {
- format!(
- "Failed to find stream with ID: {}, session: {session} (error: {error})",
- self.stream_id
+ shard.ensure_authenticated(session)?;
+ shard.ensure_topic_exists(&self.stream_id, &self.topic_id)?;
+ shard
+ .streams2
+ .with_root_by_id(&self.stream_id, |stream_root| {
+ shard.permissioner.borrow().get_topic(
+ session.get_user_id(),
+ stream_root.id() as u32,
+ self.topic_id.get_u32_value().unwrap_or(0),
)
})?;
- let Ok(topic) = shard.try_find_topic(session, &stream, &self.topic_id) else {
- sender.send_empty_ok_response().await?;
- return Ok(());
- };
- let Some(topic) = topic else {
- sender.send_empty_ok_response().await?;
- return Ok(());
- };
-
- let topic = mapper::map_topic(topic).await;
- sender.send_ok_response(&topic).await?;
+ shard
+ .streams2
+ .with_topic_by_id_async(&self.stream_id, &self.topic_id, async |(root, stats)| {
+ let response = mapper::map_topic(&root, &stats);
+ sender.send_ok_response(&response).await
+ })
+ .await?;
Ok(())
}
}
diff --git a/core/server/src/binary/handlers/topics/get_topics_handler.rs b/core/server/src/binary/handlers/topics/get_topics_handler.rs
index 1606245e..69a586a0 100644
--- a/core/server/src/binary/handlers/topics/get_topics_handler.rs
+++ b/core/server/src/binary/handlers/topics/get_topics_handler.rs
@@ -17,14 +17,13 @@
*/
use crate::binary::command::{BinaryServerCommand, ServerCommand, ServerCommandHandler};
-use crate::binary::handlers::topics::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::binary::sender::SenderKind;
use crate::shard::IggyShard;
+use crate::slab::traits_ext::{EntityComponentSystem, IntoComponents};
use crate::streaming::session::Session;
use anyhow::Result;
-use error_set::ErrContext;
use iggy_common::IggyError;
use iggy_common::get_topics::GetTopics;
use std::rc::Rc;
@@ -43,23 +42,27 @@ impl ServerCommandHandler for GetTopics {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
- let stream = shard.find_stream(session, &self.stream_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to get stream, stream_id: {}, session: {session}",
- self.stream_id
- )
+ shard.ensure_authenticated(session)?;
+ shard.ensure_stream_exists(&self.stream_id)?;
+ shard.streams2.with_root_by_id(&self.stream_id, |root| {
+ shard
+ .permissioner
+ .borrow()
+ .get_topics(session.get_user_id(), root.id() as u32)
})?;
- let topics = shard
- .find_topics(session, &stream)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to find topics, stream_id: {}, session: {session}",
- self.stream_id
- )
- })?;
- let response = mapper::map_topics(topics);
- sender.send_ok_response(&response).await?;
+
+ shard
+ .streams2
+ .with_topics_async(&self.stream_id, async |topics| {
+ topics
+ .with_async(async |topics| {
+ let (roots, stats) = topics.into_components();
+ let response = mapper::map_topics(&roots, &stats);
+ sender.send_ok_response(&response).await
+ })
+ .await
+ })
+ .await?;
Ok(())
}
}
diff --git a/core/server/src/binary/handlers/topics/update_topic_handler.rs b/core/server/src/binary/handlers/topics/update_topic_handler.rs
index e9f2a3b7..8d3acc6b 100644
--- a/core/server/src/binary/handlers/topics/update_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/update_topic_handler.rs
@@ -54,6 +54,18 @@ impl ServerCommandHandler for UpdateTopic {
self.max_topic_size,
self.replication_factor,
);
+ self.message_expiry =
+ shard
+ .streams2
+ .with_topic_root_by_id(&self.stream_id, &self.topic_id, |root| {
+ root.message_expiry()
+ });
+ self.max_topic_size =
+ shard
+ .streams2
+ .with_topic_root_by_id(&self.stream_id, &self.topic_id, |root| {
+ root.max_topic_size()
+ });
let event = ShardEvent::UpdatedTopic2 {
stream_id: self.stream_id.clone(),
topic_id: self.topic_id.clone(),
@@ -64,49 +76,6 @@ impl ServerCommandHandler for UpdateTopic {
replication_factor: self.replication_factor,
};
let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
-
- /*
- shard
- .update_topic(
- session,
- &self.stream_id,
- &self.topic_id,
- &self.name,
- self.message_expiry,
- self.compression_algorithm,
- self.max_topic_size,
- self.replication_factor,
- )
- .await
- .with_error_context(|error| format!(
- "{COMPONENT} (error: {error}) - failed to update topic with id: {}, stream_id: {}, session: {session}",
- self.topic_id, self.stream_id
- ))?;
- let event = ShardEvent::UpdatedTopic {
- stream_id: self.stream_id.clone(),
- topic_id: self.topic_id.clone(),
- name: self.name.clone(),
- message_expiry: self.message_expiry,
- compression_algorithm: self.compression_algorithm,
- max_topic_size: self.max_topic_size,
- replication_factor: self.replication_factor,
- };
- let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
-
- let stream = shard.find_stream(session, &self.stream_id)
- .with_error_context(|error| format!(
- "{COMPONENT} (error: {error}) - failed to find stream, stream_id: {}, session: {session}",
- self.stream_id
- ))?;
- let topic = stream
- .get_topic(&self.topic_id)
- .with_error_context(|error| format!(
- "{COMPONENT} (error: {error}) - failed to find topic, topic_id: {}, stream_id: {}, session: {session}",
- self.topic_id, self.stream_id
- ))?;
- self.message_expiry = topic.message_expiry;
- self.max_topic_size = topic.max_topic_size;
-
let topic_id = self.topic_id.clone();
let stream_id = self.stream_id.clone();
@@ -117,7 +86,6 @@ impl ServerCommandHandler for UpdateTopic {
.with_error_context(|error| format!(
"{COMPONENT} (error: {error}) - failed to apply update topic with id: {topic_id}, stream_id: {stream_id}, session: {session}"
))?;
- */
sender.send_empty_ok_response().await?;
Ok(())
}
diff --git a/core/server/src/binary/mapper.rs b/core/server/src/binary/mapper.rs
index 8caf9d1f..0b2ced4d 100644
--- a/core/server/src/binary/mapper.rs
+++ b/core/server/src/binary/mapper.rs
@@ -17,18 +17,22 @@
*/
use std::cell::Ref;
+use std::sync::{Arc, atomic::AtomicU64};
+use crate::slab::traits_ext::{EntityComponentSystem, IntoComponents};
use crate::streaming::clients::client_manager::{Client, Transport};
-use crate::streaming::partitions::partition::Partition;
+use crate::streaming::partitions::partition2::PartitionRoot;
use crate::streaming::personal_access_tokens::personal_access_token::PersonalAccessToken;
+use crate::streaming::stats::stats::{PartitionStats, StreamStats, TopicStats};
use crate::streaming::streams::stream::Stream;
use crate::streaming::streams::stream2;
use crate::streaming::topics::consumer_group::ConsumerGroup;
use crate::streaming::topics::topic::Topic;
+use crate::streaming::topics::topic2::{self, TopicRoot};
use crate::streaming::users::user::User;
use bytes::{BufMut, Bytes, BytesMut};
-use iggy_common::locking::IggyRwLockFn;
-use iggy_common::{BytesSerializable, ConsumerOffsetInfo, Sizeable, Stats, UserId};
+use iggy_common::{BytesSerializable, ConsumerOffsetInfo, Stats, UserId};
+use slab::Slab;
pub fn map_stats(stats: &Stats) -> Bytes {
let mut bytes = BytesMut::with_capacity(104);
@@ -149,48 +153,53 @@ pub fn map_personal_access_tokens(personal_access_tokens: Vec<PersonalAccessToke
bytes.freeze()
}
-pub fn map_stream2(stream: &stream2::Stream) -> Bytes {
+pub fn map_streams(roots: &Slab<stream2::StreamRoot>, stats: &Slab<Arc<StreamStats>>) -> Bytes {
let mut bytes = BytesMut::new();
- extend_stream2(stream, &mut bytes);
- /*
- for topic in stream.get_topics() {
- extend_topic(topic, &mut bytes);
- }
- */
- bytes.freeze()
-}
-pub fn map_stream(stream: &Stream) -> Bytes {
- let mut bytes = BytesMut::new();
- extend_stream(stream, &mut bytes);
- for topic in stream.get_topics() {
- extend_topic(topic, &mut bytes);
+ for (root, stat) in roots
+ .iter()
+ .map(|(_, val)| val)
+ .zip(stats.iter().map(|(_, val)| val))
+ {
+ extend_stream(root, stat, &mut bytes);
}
bytes.freeze()
}
-pub fn map_streams(streams: Vec<Ref<'_, Stream>>) -> Bytes {
+pub fn map_stream(root: &stream2::StreamRoot, stats: &StreamStats) -> Bytes {
let mut bytes = BytesMut::new();
- for stream in streams.iter() {
- extend_stream(stream, &mut bytes);
- }
+ extend_stream(root, stats, &mut bytes);
bytes.freeze()
}
-pub fn map_topics(topics: Vec<&Topic>) -> Bytes {
+pub fn map_topics(roots: &Slab<TopicRoot>, stats: &Slab<Arc<TopicStats>>) -> Bytes {
let mut bytes = BytesMut::new();
- for topic in topics {
- extend_topic(topic, &mut bytes);
+ for (root, stat) in roots
+ .iter()
+ .map(|(_, val)| val)
+ .zip(stats.iter().map(|(_, val)| val))
+ {
+ extend_topic(root, stat, &mut bytes);
}
bytes.freeze()
}
-pub async fn map_topic(topic: &Topic) -> Bytes {
+pub fn map_topic(root: &topic2::TopicRoot, stats: &TopicStats) -> Bytes {
let mut bytes = BytesMut::new();
- extend_topic(topic, &mut bytes);
- for partition in topic.get_partitions() {
- let partition = partition.read().await;
- extend_partition(&partition, &mut bytes);
- }
+ extend_topic(root, stats, &mut bytes);
+
+ root.partitions().with(|partitions| {
+ let (roots, stats, _, offsets, _, _) = partitions.into_components();
+ for (root, stat, offset) in roots
+ .iter()
+ .map(|(_, val)| val)
+ .zip(stats.iter().map(|(_, val)| val))
+ .zip(offsets.iter().map(|(_, val)| val))
+ .map(|((root, stat), offset)| (root, stat, offset))
+ {
+ extend_partition(root, stat, offset, &mut bytes);
+ }
+ });
+
bytes.freeze()
}
@@ -217,12 +226,17 @@ pub fn map_consumer_groups(consumer_groups: Vec<ConsumerGroup>) -> Bytes {
bytes.freeze()
}
-fn extend_stream2(stream: &stream2::Stream, bytes: &mut BytesMut) {
- //stream.serialize_into(bytes);
- // TODO:
+fn extend_stream(root: &stream2::StreamRoot, stats: &StreamStats, bytes: &mut BytesMut) {
+ bytes.put_u32_le(root.id() as u32);
+ bytes.put_u64_le(root.created_at().into());
+ bytes.put_u32_le(root.topics_count() as u32);
+ bytes.put_u64_le(stats.size_bytes_inconsistent());
+ bytes.put_u64_le(stats.messages_count_inconsistent());
+ bytes.put_u8(root.name().len() as u8);
+ bytes.put_slice(root.name().as_bytes());
}
-fn extend_stream(stream: &Stream, bytes: &mut BytesMut) {
+fn extend_stream_legacy(stream: &Stream, bytes: &mut BytesMut) {
bytes.put_u32_le(stream.stream_id);
bytes.put_u64_le(stream.created_at.into());
bytes.put_u32_le(stream.get_topics().len() as u32);
@@ -232,27 +246,50 @@ fn extend_stream(stream: &Stream, bytes: &mut BytesMut) {
bytes.put_slice(stream.name.as_bytes());
}
-fn extend_topic(topic: &Topic, bytes: &mut BytesMut) {
+fn extend_topic_legacy(topic: &Topic, bytes: &mut BytesMut) {
bytes.put_u32_le(topic.topic_id);
bytes.put_u64_le(topic.created_at.into());
- bytes.put_u32_le(topic.get_partitions().len() as u32);
+ bytes.put_u32_le(topic.partitions.len() as u32);
bytes.put_u64_le(topic.message_expiry.into());
bytes.put_u8(topic.compression_algorithm.as_code());
bytes.put_u64_le(topic.max_topic_size.into());
bytes.put_u8(topic.replication_factor);
- bytes.put_u64_le(topic.get_size_bytes().as_bytes_u64());
- bytes.put_u64_le(topic.get_messages_count());
+ bytes.put_u64_le(topic.size_bytes.load(std::sync::atomic::Ordering::Relaxed));
+ bytes.put_u64_le(
+ topic
+ .messages_count
+ .load(std::sync::atomic::Ordering::Relaxed),
+ );
bytes.put_u8(topic.name.len() as u8);
bytes.put_slice(topic.name.as_bytes());
}
-fn extend_partition(partition: &Partition, bytes: &mut BytesMut) {
- bytes.put_u32_le(partition.partition_id);
- bytes.put_u64_le(partition.created_at.into());
- bytes.put_u32_le(partition.get_segments().len() as u32);
- bytes.put_u64_le(partition.current_offset);
- bytes.put_u64_le(partition.get_size_bytes().as_bytes_u64());
- bytes.put_u64_le(partition.get_messages_count());
+fn extend_topic(root: &TopicRoot, stats: &TopicStats, bytes: &mut BytesMut) {
+ bytes.put_u32_le(root.id() as u32);
+ bytes.put_u64_le(root.created_at().into());
+ bytes.put_u32_le(root.partitions().len() as u32);
+ bytes.put_u64_le(root.message_expiry().into());
+ bytes.put_u8(root.compression_algorithm().as_code());
+ bytes.put_u64_le(root.max_topic_size().into());
+ bytes.put_u8(root.replication_factor());
+ bytes.put_u64_le(stats.size_bytes_inconsistent());
+ bytes.put_u64_le(stats.messages_count_inconsistent());
+ bytes.put_u8(root.name().len() as u8);
+ bytes.put_slice(root.name().as_bytes());
+}
+
+fn extend_partition(
+ root: &PartitionRoot,
+ stats: &PartitionStats,
+ offset: &Arc<AtomicU64>,
+ bytes: &mut BytesMut,
+) {
+ bytes.put_u32_le(root.id() as u32);
+ bytes.put_u64_le(root.created_at().into());
+ bytes.put_u32_le(stats.segments_count_inconsistent());
+ bytes.put_u64_le(offset.load(std::sync::atomic::Ordering::Relaxed));
+ bytes.put_u64_le(stats.size_bytes_inconsistent());
+ bytes.put_u64_le(stats.messages_count_inconsistent());
}
fn extend_consumer_group(consumer_group: &ConsumerGroup, bytes: &mut BytesMut) {
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index e7cb032a..6f06d4ad 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -559,16 +559,7 @@ impl IggyShard {
replication_factor,
} => {
let topic_id = topic_id.get_u32_value().ok();
- self.create_topic_bypass_auth(
- &stream_id,
- None,
- &name,
- partitions_count,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor,
- )
+ Ok(())
}
ShardEvent::LoginUser {
client_id,
@@ -730,19 +721,7 @@ impl IggyShard {
compression_algorithm,
max_topic_size,
replication_factor,
- } => {
- self.update_topic_bypass_auth(
- &stream_id,
- &topic_id,
- &name,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor,
- )
- .await?;
- Ok(())
- }
+ } => Ok(()),
ShardEvent::PurgedStream { stream_id } => {
self.purge_stream_bypass_auth(&stream_id).await?;
Ok(())
@@ -761,16 +740,7 @@ impl IggyShard {
ShardEvent::DeletedTopic {
stream_id,
topic_id,
- } => {
- self.delete_topic_bypass_auth(&stream_id, &topic_id)
- .await
- .with_error_context(|err| {
- format!(
- "{COMPONENT} (error: {err}) - failed to delete topic with ID: {topic_id} in stream with ID: {stream_id}"
- )
- })?;
- Ok(())
- }
+ } => Ok(()),
ShardEvent::CreatedConsumerGroup {
stream_id,
topic_id,
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index 3cc1ed61..dfd9efe4 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -176,8 +176,8 @@ impl IggyShard {
partition: partition2::Partition,
) -> usize {
self.streams2
- .with_partitions_mut(stream_id, topic_id, |partitions| {
- partitions.insert(partition)
+ .with_topic_root_by_id_mut(stream_id, topic_id, |root| {
+ root.partitions_mut().insert(partition)
})
}
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index 54e639fd..908d81aa 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -19,202 +19,27 @@
use super::COMPONENT;
use crate::shard::IggyShard;
use crate::shard::namespace::IggyNamespace;
-use crate::slab::traits_ext::{
- DeleteCell, EntityComponentSystem, EntityMarker, Insert, InsertCell, IntoComponents,
-};
-use crate::streaming::partitions::partition;
+use crate::slab::traits_ext::{DeleteCell, EntityMarker, InsertCell};
+
use crate::streaming::session::Session;
use crate::streaming::stats::stats::StreamStats;
use crate::streaming::streams::storage2::create_stream_file_hierarchy;
use crate::streaming::streams::stream::Stream;
-use crate::streaming::streams::stream2;
+use crate::streaming::streams::{self, stream2};
use error_set::ErrContext;
-use futures::future::try_join_all;
+
use iggy_common::locking::IggyRwLockFn;
use iggy_common::{IdKind, Identifier, IggyError, IggyTimestamp};
-use std::cell::{Ref, RefCell, RefMut};
+use std::cell::{Ref, RefMut};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
-use tokio::fs;
-use tracing::{error, info, warn};
-thread_local! {
- static CURRENT_STREAM_ID: AtomicU32 = AtomicU32::new(1);
-}
impl IggyShard {
- pub fn get_streams(&self) -> Vec<Ref<'_, Stream>> {
- let len = self.streams.borrow().len();
- let result = (0..len)
- .map(|i| {
- Ref::map(self.streams.borrow(), |streams| {
- streams.values().nth(i).unwrap()
- })
- })
- .collect();
- result
- }
-
- pub fn find_streams(&self, session: &Session) -> Result<Vec<Ref<'_, Stream>>, IggyError> {
- self.ensure_authenticated(session)?;
- self.permissioner
- .borrow()
- .get_streams(session.get_user_id())
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to get streams for user {}",
- session.get_user_id(),
- )
- })?;
- Ok(self.get_streams())
- }
-
- pub fn find_stream(
- &self,
- session: &Session,
- identifier: &Identifier,
- ) -> Result<Ref<'_, Stream>, IggyError> {
- self.ensure_authenticated(session)?;
- let stream = self.get_stream(identifier);
- if let Ok(stream) = stream {
- self.permissioner
- .borrow()
- .get_stream(session.get_user_id(), stream.stream_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to get stream for user {}",
- session.get_user_id(),
- )
- })?;
- return Ok(stream);
- }
-
- stream
- }
-
- pub fn try_find_stream(
- &self,
- session: &Session,
- identifier: &Identifier,
- ) -> Result<Option<Ref<'_, Stream>>, IggyError> {
- self.ensure_authenticated(session)?;
- let Some(stream) = self.try_get_stream(identifier)? else {
- return Ok(None);
- };
-
- self.permissioner
- .borrow()
- .get_stream(session.get_user_id(), stream.stream_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to get stream with ID: {identifier} for user with ID: {}",
- session.get_user_id(),
- )
- })?;
- Ok(Some(stream))
- }
-
- pub fn try_get_stream(
- &self,
- identifier: &Identifier,
- ) -> Result<Option<Ref<'_, Stream>>, IggyError> {
- match identifier.kind {
- IdKind::Numeric => self.get_stream_by_id(identifier.get_u32_value()?).map(Some),
- IdKind::String => Ok(self.try_get_stream_by_name(&identifier.get_cow_str_value()?)),
- }
- }
-
- pub fn try_get_topic_id(&self, stream_id: &Identifier, name: &str) -> Option<u32> {
- let stream = self.get_stream(stream_id).ok()?;
- stream.topics_ids.get(name).and_then(|id| Some(*id))
- }
-
- pub fn try_get_stream_id(&self, name: &str) -> Option<u32> {
- self.streams_ids.borrow().get(name).and_then(|id| Some(*id))
- }
-
- fn try_get_stream_by_name(&self, name: &str) -> Option<Ref<'_, Stream>> {
- self.streams_ids
- .borrow()
- .get(name)
- .and_then(|id| Some(self.get_stream_ref(*id)))
- }
-
- pub fn get_stream(&self, identifier: &Identifier) -> Result<Ref<'_, Stream>, IggyError> {
- match identifier.kind {
- IdKind::Numeric => self.get_stream_by_id(identifier.get_u32_value()?),
- IdKind::String => self.get_stream_by_name(&identifier.get_cow_str_value()?),
- }
- }
-
- pub fn get_stream_mut(&self, identifier: &Identifier) -> Result<RefMut<'_, Stream>, IggyError> {
- match identifier.kind {
- IdKind::Numeric => self.get_stream_by_id_mut(identifier.get_u32_value()?),
- IdKind::String => self.get_stream_by_name_mut(&identifier.get_cow_str_value()?),
- }
- }
-
- fn get_stream_by_name(&self, name: &str) -> Result<Ref<'_, Stream>, IggyError> {
- let exists = self.streams_ids.borrow().iter().any(|s| s.0 == name);
- if !exists {
- return Err(IggyError::StreamNameNotFound(name.to_string()));
- }
- let stream_id = self.streams_ids.borrow().get(name).cloned().unwrap();
- self.get_stream_by_id(stream_id)
- }
-
- fn get_stream_by_id(&self, stream_id: u32) -> Result<Ref<'_, Stream>, IggyError> {
- let exists = self.streams.borrow().iter().any(|s| s.0 == &stream_id);
- if !exists {
- return Err(IggyError::StreamIdNotFound(stream_id));
- }
- Ok(self.get_stream_ref(stream_id))
- }
-
- fn get_stream_ref(&self, stream_id: u32) -> Ref<'_, Stream> {
- Ref::map(self.streams.borrow(), |streams| {
- streams.get(&stream_id).expect("Stream ID not found")
- })
- }
-
- fn get_stream_by_name_mut(&self, name: &str) -> Result<RefMut<'_, Stream>, IggyError> {
- let exists = self.streams_ids.borrow().iter().any(|s| s.0 == name);
- if !exists {
- return Err(IggyError::StreamNameAlreadyExists(name.to_owned()));
- }
- let streams_ids = self.streams_ids.borrow();
- let id = streams_ids.get(name).cloned();
- drop(streams_ids);
- self.get_stream_by_id_mut(id.unwrap())
- }
-
- fn get_stream_by_id_mut(&self, stream_id: u32) -> Result<RefMut<'_, Stream>, IggyError> {
- let exists = self.streams.borrow().iter().any(|s| s.0 == &stream_id);
- if !exists {
- return Err(IggyError::StreamIdNotFound(stream_id));
- }
- Ok(RefMut::map(self.streams.borrow_mut(), |s| {
- s.get_mut(&stream_id).unwrap()
- }))
- }
-
- pub fn create_stream_bypass_auth(
- &self,
- stream_id: Option<u32>,
- name: &str,
- ) -> Result<u32, IggyError> {
- let stream = self.create_stream_base(stream_id, name)?;
- let id = stream.stream_id;
- self.streams_ids.borrow_mut().insert(name.to_owned(), id);
- self.streams.borrow_mut().insert(id, stream);
- self.metrics.increment_streams(1);
- Ok(id)
- }
-
pub async fn create_stream2(
&self,
session: &Session,
- stream_id: Option<u32>,
+ _stream_id: Option<u32>,
name: String,
) -> Result<stream2::Stream, IggyError> {
self.ensure_authenticated(session)?;
@@ -250,145 +75,6 @@ impl IggyShard {
self.insert_stream_mem(stream)
}
- pub async fn create_stream(
- &self,
- session: &Session,
- stream_id: Option<u32>,
- name: &str,
- ) -> Result<Identifier, IggyError> {
- self.ensure_authenticated(session)?;
- self.permissioner
- .borrow()
- .create_stream(session.get_user_id())?;
- if self.streams_ids.borrow().contains_key(name) {
- return Err(IggyError::StreamNameAlreadyExists(name.to_owned()));
- }
- let stream = self
- .create_stream_base(stream_id, name)
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to create stream with name: {name}")
- })?;
- let id = stream.stream_id;
-
- stream.persist().await?;
-
- self.streams_ids.borrow_mut().insert(name.to_owned(), id);
- self.streams.borrow_mut().insert(id, stream);
- self.metrics.increment_streams(1);
- Ok(Identifier::numeric(id)?)
- }
-
- fn create_stream_base(&self, stream_id: Option<u32>, name: &str) -> Result<Stream, IggyError> {
- if self.streams_ids.borrow().contains_key(name) {
- return Err(IggyError::StreamNameAlreadyExists(name.to_owned()));
- }
-
- let mut id;
- if stream_id.is_none() {
- id = CURRENT_STREAM_ID.with(|current_id| current_id.fetch_add(1, Ordering::SeqCst));
- loop {
- if self.streams.borrow().contains_key(&id) {
- if id == u32::MAX {
- return Err(IggyError::StreamIdAlreadyExists(id));
- }
- id = CURRENT_STREAM_ID
- .with(|current_id| current_id.fetch_add(1, Ordering::SeqCst));
- } else {
- break;
- }
- }
- } else {
- id = stream_id.unwrap();
- }
-
- if self.streams.borrow().contains_key(&id) {
- return Err(IggyError::StreamIdAlreadyExists(id));
- }
-
- let stream = Stream::create(id, name, self.config.system.clone(), self.storage.clone());
- Ok(stream)
- }
-
- pub fn update_stream_bypass_auth(&self, id: &Identifier, name: &str) -> Result<(), IggyError> {
- let stream_id;
- {
- let stream = self.get_stream(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {id}")
- })?;
- stream_id = stream.stream_id;
- }
-
- let old_name = self.update_stream_base(stream_id, id, name)?;
- info!("Stream with ID '{id}' updated. Old name: '{old_name}' changed to: '{name}'.");
- Ok(())
- }
-
- pub async fn update_stream(
- &self,
- session: &Session,
- id: &Identifier,
- name: &str,
- ) -> Result<(), IggyError> {
- self.ensure_authenticated(session)?;
- let stream_id;
- {
- let stream = self.get_stream(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {id}")
- })?;
- stream_id = stream.stream_id;
- }
-
- self.permissioner
- .borrow()
- .update_stream(session.get_user_id(), stream_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to update stream, user ID: {}, stream ID: {}",
- session.get_user_id(),
- stream_id
- )
- })?;
-
- let old_name = self.update_stream_base(stream_id, id, name)?;
- let stream = self.get_stream(id)?;
- stream.persist().await.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to persist stream with ID: {stream_id}, name: {name}")
- })?;
-
- info!("Stream with ID '{id}' updated. Old name: '{old_name}' changed to: '{name}'.");
- Ok(())
- }
-
- fn update_stream_base(
- &self,
- stream_id: u32,
- id: &Identifier,
- name: &str,
- ) -> Result<String, IggyError> {
- if let Some(stream_id_by_name) = self.streams_ids.borrow().get(name) {
- if *stream_id_by_name != stream_id {
- return Err(IggyError::StreamNameAlreadyExists(name.to_owned()));
- }
- }
- let old_name;
- {
- let mut stream = self.get_stream_mut(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get mutable reference to stream with id: {id}")
- })?;
- old_name = stream.name.clone();
- stream.name = name.to_owned();
- // Drop the exclusive borrow.
- }
-
- {
- self.streams_ids.borrow_mut().remove(&old_name);
- self.streams_ids
- .borrow_mut()
- .insert(name.to_owned(), stream_id);
- }
- Ok(old_name)
- }
-
pub fn update_stream2_bypass_auth(&self, id: &Identifier, name: &str) -> Result<(), IggyError> {
self.update_stream2_base(id, name.to_string())?;
Ok(())
@@ -397,15 +83,16 @@ impl IggyShard {
pub fn update_stream2(
&self,
session: &Session,
- id: &Identifier,
- name: &str,
+ stream_id: &Identifier,
+ name: String,
) -> Result<(), IggyError> {
self.ensure_authenticated(session)?;
- let stream_id = self.streams2.with_root_by_id(id, |root| root.id() as u32);
+ self.ensure_stream_exists(stream_id)?;
+ let id = self.streams2.with_stream_by_id(stream_id, streams::helpers::get_stream_id());
self.permissioner
.borrow()
- .update_stream(session.get_user_id(), stream_id)
+ .update_stream(session.get_user_id(), id as u32)
.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - failed to update stream, user ID: {}, stream ID: {}",
@@ -413,95 +100,29 @@ impl IggyShard {
stream_id
)
})?;
- self.update_stream2_base(id, name.to_string())?;
+ self.update_stream2_base(stream_id, name)?;
Ok(())
}
- fn update_stream2_base(&self, id: &Identifier, name: String) -> Result<String, IggyError> {
+ fn update_stream2_base(&self, id: &Identifier, name: String) -> Result<(), IggyError> {
let old_name = self
.streams2
- .with_root_by_id(id, |root| root.name().clone());
+ .with_stream_by_id(id, streams::helpers::get_stream_name());
if old_name == name {
- return Ok(old_name);
+ return Ok(());
}
-
- // if stream with name already exists, return error
if self.streams2.with_index(|index| index.contains_key(&name)) {
return Err(IggyError::StreamNameAlreadyExists(name.to_string()));
}
- self.streams2.with_root_by_id_mut(id, |root| {
- root.set_name(name.clone());
- });
-
+ self.streams2.with_stream_by_id_mut(id, streams::helpers::update_stream_name(name.clone()));
self.streams2.with_index_mut(|index| {
// Rename the key inside of hashmap
let idx = index.remove(&old_name).expect("Rename key: key not found");
index.insert(name, idx);
});
- Ok(old_name)
- }
-
- pub fn delete_stream_bypass_auth(&self, id: &Identifier) -> Result<Stream, IggyError> {
- let stream_id;
- let stream_name;
- {
- let stream = self.get_stream(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {id}")
- })?;
- stream_id = stream.stream_id;
- stream_name = stream.name.clone();
- }
- let stream = self.delete_stream_base(stream_id, stream_name)?;
- Ok(stream)
- }
-
- pub fn delete_stream(&self, session: &Session, id: &Identifier) -> Result<Stream, IggyError> {
- self.ensure_authenticated(session)?;
- let stream_id;
- let stream_name;
- {
- let stream = self.get_stream(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {id}")
- })?;
- stream_id = stream.stream_id;
- stream_name = stream.name.clone();
- }
- self.permissioner
- .borrow()
- .delete_stream(session.get_user_id(), stream_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to delete stream for user {}, stream ID: {}",
- session.get_user_id(),
- stream_id,
- )
- })?;
- let stream = self.delete_stream_base(stream_id, stream_name)?;
- Ok(stream)
- }
-
- fn delete_stream_base(&self, stream_id: u32, stream_name: String) -> Result<Stream, IggyError> {
- let stream = self.streams.borrow_mut().remove(&stream_id).unwrap();
- self.streams_ids.borrow_mut().remove(&stream_name);
-
- self.metrics.decrement_streams(1);
- self.metrics.decrement_topics(stream.get_topics_count());
- self.metrics
- .decrement_partitions(stream.get_partitions_count());
- self.metrics.decrement_messages(stream.get_messages_count());
- self.metrics.decrement_segments(stream.get_segments_count());
- let current_stream_id =
- CURRENT_STREAM_ID.with(|current_id| current_id.load(Ordering::SeqCst));
- if current_stream_id > stream_id {
- CURRENT_STREAM_ID.with(|current_id| current_id.store(stream_id, Ordering::SeqCst));
- }
-
- self.client_manager
- .borrow_mut()
- .delete_consumer_groups_for_stream(stream_id);
- Ok(stream)
+ Ok(())
}
pub fn delete_stream2_bypass_auth(
@@ -520,8 +141,10 @@ impl IggyShard {
self.metrics.decrement_streams(1);
self.metrics.decrement_topics(0); // TODO: stats doesn't have topic count
self.metrics.decrement_partitions(0); // TODO: stats doesn't have partition count
- self.metrics.decrement_messages(stats.messages_count());
- self.metrics.decrement_segments(stats.segments_count());
+ self.metrics
+ .decrement_messages(stats.messages_count_inconsistent());
+ self.metrics
+ .decrement_segments(stats.segments_count_inconsistent());
/*
self.client_manager
@@ -537,8 +160,8 @@ impl IggyShard {
id: &Identifier,
) -> Result<stream2::Stream, IggyError> {
self.ensure_authenticated(session)?;
- // self.ensure_stream_exists(id)?;
- let stream_id = self.streams2.with_root_by_id(id, |root| root.id() as u32);
+ self.ensure_stream_exists(id)?;
+ let stream_id = self.streams2.with_stream_by_id(id, streams::helpers::get_stream_id());
self.permissioner
.borrow()
.delete_stream(session.get_user_id(), stream_id as u32)
@@ -553,60 +176,14 @@ impl IggyShard {
Ok(stream)
}
- pub async fn purge_stream(
- &self,
- session: &Session,
- stream_id: &Identifier,
- ) -> Result<(), IggyError> {
- let stream = self.get_stream(stream_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
- })?;
- self.permissioner
- .borrow()
- .purge_stream(session.get_user_id(), stream.stream_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to purge stream for user {}, stream ID: {}",
- session.get_user_id(),
- stream.stream_id,
- )
- })?;
- self.purge_stream_base(stream.stream_id).await?;
- Ok(())
- }
-
- pub async fn purge_stream_bypass_auth(&self, stream_id: &Identifier) -> Result<(), IggyError> {
- let stream = self.get_stream(stream_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
- })?;
- self.purge_stream_base(stream.stream_id).await?;
- Ok(())
- }
-
- async fn purge_stream_base(&self, stream_id: u32) -> Result<(), IggyError> {
- let stream = self.get_stream_ref(stream_id);
- for topic in stream.get_topics() {
- let topic_id = topic.topic_id;
- for partition in topic.get_partitions() {
- let mut partition = partition.write().await;
- let partition_id = partition.partition_id;
- let namespace = IggyNamespace::new(stream_id, topic_id, partition_id);
- let shard_info = self.find_shard_table_record(&namespace).unwrap();
- if shard_info.id() == self.id {
- partition.purge().await?;
- }
- }
- }
- Ok(())
- }
-
pub fn purge_stream2(&self, session: &Session, id: &Identifier) -> Result<(), IggyError> {
self.ensure_authenticated(session)?;
// self.ensure_stream_exists(id)?;
- let stream_id = self.streams2.with_root_by_id(id, |root| root.id() as u32);
+ let get_stream_id = crate::streaming::streams::helpers::get_stream_id();
+ let stream_id = self.streams2.with_stream_by_id(id, get_stream_id);
self.permissioner
.borrow()
- .purge_stream(session.get_user_id(), stream_id)
+ .purge_stream(session.get_user_id(), stream_id as u32)
.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - permission denied to purge stream for user {}, stream ID: {}",
@@ -623,7 +200,7 @@ impl IggyShard {
Ok(())
}
- fn purge_stream2_base(&self, stream_id: &Identifier) -> Result<(), IggyError> {
+ fn purge_stream2_base(&self, _stream_id: &Identifier) -> Result<(), IggyError> {
// TODO
Ok(())
}
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index d0beaa31..4daf0545 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -16,143 +16,31 @@
* under the License.
*/
-use std::str::FromStr;
-use std::sync::Arc;
-
use super::COMPONENT;
+use crate::shard::IggyShard;
use crate::shard::namespace::IggyNamespace;
-use crate::shard::transmission::event::ShardEvent;
-use crate::shard::{IggyShard, ShardInfo};
use crate::shard_info;
-use crate::slab::traits_ext::{
- Delete, DeleteCell, EntityComponentSystem, EntityMarker, Insert, InsertCell,
-};
-use crate::state::system::StreamState;
-use crate::streaming::partitions::partition2;
-use crate::streaming::partitions::storage2::create_partition_file_hierarchy;
+use crate::slab::traits_ext::{EntityMarker, InsertCell};
use crate::streaming::session::Session;
use crate::streaming::stats::stats::{StreamStats, TopicStats};
use crate::streaming::streams::stream::Stream;
use crate::streaming::topics::storage2::create_topic_file_hierarchy;
use crate::streaming::topics::topic::Topic;
-use crate::streaming::topics::topic2;
-use clap::Id;
+use crate::streaming::topics::topic2::{self, resolve_max_topic_size, resolve_message_expiry};
+use crate::streaming::{streams, topics};
use error_set::ErrContext;
use iggy_common::locking::IggyRwLockFn;
use iggy_common::{
CompressionAlgorithm, Identifier, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize,
};
-use tokio_util::io::StreamReader;
-use tracing::info;
+use std::str::FromStr;
+use std::sync::Arc;
impl IggyShard {
- pub fn find_topic<'topic, 'stream>(
- &self,
- session: &Session,
- stream: &'stream Stream,
- topic_id: &Identifier,
- ) -> Result<&'topic Topic, IggyError>
- where
- 'stream: 'topic,
- {
- self.ensure_authenticated(session)?;
- let stream_id = stream.stream_id;
- let topic = stream.get_topic(topic_id)?;
- self.permissioner
- .borrow()
- .get_topic(session.get_user_id(), stream.stream_id, topic.topic_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to get topic with ID: {topic_id} in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
- )
- })?;
- Ok(topic)
- }
-
- pub fn find_topics<'stream, 'topic>(
- &self,
- session: &Session,
- stream: &'stream Stream,
- ) -> Result<Vec<&'topic Topic>, IggyError>
- where
- 'stream: 'topic,
- {
- self.ensure_authenticated(session)?;
- let stream_id = stream.stream_id;
- self.permissioner
- .borrow()
- .get_topics(session.get_user_id(), stream.stream_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to get topics in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
- )
- })?;
- Ok(stream.get_topics())
- }
-
- pub fn try_find_topic<'stream, 'topic>(
- &self,
- session: &Session,
- stream: &'stream Stream,
- topic_id: &Identifier,
- ) -> Result<Option<&'topic Topic>, IggyError>
- where
- 'stream: 'topic,
- {
- self.ensure_authenticated(session)?;
- let stream_id = stream.stream_id;
- let Some(topic) = stream.try_get_topic(topic_id)? else {
- return Ok(None);
- };
-
- self.permissioner
- .borrow()
- .get_topic(session.get_user_id(), stream_id, topic.topic_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to get topic with ID: {topic_id} in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
- )
- })?;
- Ok(Some(topic))
- }
-
- pub fn create_topic_bypass_auth(
- &self,
- stream_id: &Identifier,
- topic_id: Option<u32>,
- name: &str,
- partitions_count: u32,
- message_expiry: IggyExpiry,
- compression_algorithm: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- replication_factor: Option<u8>,
- ) -> Result<(), IggyError> {
- self.create_topic_base(
- stream_id,
- topic_id,
- name,
- partitions_count,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor,
- )?;
-
- self.metrics.increment_topics(1);
- self.metrics.increment_partitions(partitions_count);
- self.metrics.increment_segments(partitions_count);
- Ok(())
- }
-
- #[allow(clippy::too_many_arguments)]
pub async fn create_topic2(
&self,
session: &Session,
stream_id: &Identifier,
- topic_id: Option<u32>,
name: String,
message_expiry: IggyExpiry,
compression: CompressionAlgorithm,
@@ -161,7 +49,7 @@ impl IggyShard {
) -> Result<topic2::Topic, IggyError> {
self.ensure_authenticated(session)?;
//self.ensure_stream_exists(stream_id)?;
- let numeric_stream_id = self.streams2.with_root_by_id(stream_id, |root| root.id());
+ let numeric_stream_id = self.streams2.get_index(stream_id);
{
self.permissioner
.borrow()
@@ -173,9 +61,10 @@ impl IggyShard {
)
})?;
}
- let exists = self.streams2.with_topics(stream_id, |topics| {
- topics.exists(&Identifier::from_str_value(&name).unwrap())
- });
+ let exists = self.streams2.with_topics(
+ stream_id,
+ topics::helpers::exists(&Identifier::from_str(&name).unwrap()),
+ );
if exists {
return Err(IggyError::TopicNameAlreadyExists(
name,
@@ -183,9 +72,13 @@ impl IggyShard {
));
}
+ let config = &self.config.system;
let parent_stats = self
.streams2
- .with_stats_by_id(stream_id, |stats| stats.clone());
+ .with_stream_by_id(stream_id, |(_, stats)| stats.clone());
+ let message_expiry = resolve_message_expiry(message_expiry, config);
+ shard_info!(self.id, "Topic message expiry: {}", message_expiry);
+ let max_topic_size = resolve_max_topic_size(max_topic_size, config)?;
let topic = self.create_and_insert_topics_mem(
stream_id,
name,
@@ -195,11 +88,6 @@ impl IggyShard {
max_topic_size,
parent_stats,
);
- let message_expiry = match topic.root().message_expiry() {
- IggyExpiry::ServerDefault => self.config.system.segment.message_expiry,
- _ => message_expiry,
- };
- shard_info!(self.id, "Topic message expiry: {}", message_expiry);
// Create file hierarchy for the topic.
create_topic_file_hierarchy(self.id, numeric_stream_id, topic.id(), &self.config.system)
@@ -236,7 +124,7 @@ impl IggyShard {
fn insert_topic_mem(&self, stream_id: &Identifier, topic: topic2::Topic) -> usize {
self.streams2
- .with_root_by_id(stream_id, |root| root.topics().insert(topic))
+ .with_topics(stream_id, |topics| topics.insert(topic))
}
pub fn create_topic2_bypass_auth(
@@ -248,119 +136,6 @@ impl IggyShard {
Ok(topic_id)
}
- #[allow(clippy::too_many_arguments)]
- pub async fn create_topic(
- &self,
- session: &Session,
- stream_id: &Identifier,
- topic_id: Option<u32>,
- name: &str,
- partitions_count: u32,
- message_expiry: IggyExpiry,
- compression_algorithm: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- replication_factor: Option<u8>,
- ) -> Result<(Identifier, Vec<u32>), IggyError> {
- self.ensure_authenticated(session)?;
- {
- let stream = self.get_stream(stream_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
- })?;
- self.permissioner
- .borrow()
- .create_topic(session.get_user_id(), stream.stream_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to create topic with name: {name} in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
- )
- })?;
- }
-
- let (topic_id, partition_ids) = self.create_topic_base(
- stream_id,
- topic_id,
- name,
- partitions_count,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor,
- )?;
-
- let stream = self.get_stream(stream_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
- })?;
- let numeric_stream_id = stream.stream_id;
- let topic = stream
- .get_topic(&topic_id)
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id} in stream with ID: {numeric_stream_id}")
- })?;
- topic.persist().await.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to persist topic: {topic}")
- })?;
-
- self.metrics.increment_topics(1);
- self.metrics.increment_partitions(partitions_count);
- self.metrics.increment_segments(partitions_count);
-
- Ok((topic_id, partition_ids))
- }
-
- fn create_topic_base(
- &self,
- stream_id: &Identifier,
- topic_id: Option<u32>,
- name: &str,
- partitions_count: u32,
- message_expiry: IggyExpiry,
- compression_algorithm: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- replication_factor: Option<u8>,
- ) -> Result<(Identifier, Vec<u32>), IggyError> {
- let mut stream = self.get_stream_mut(stream_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get mutable reference to stream with ID: {stream_id}")
- })?;
- let stream_id = stream.stream_id;
- let (topic_id, partition_ids) = stream
- .create_topic(
- topic_id,
- name,
- partitions_count,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor.unwrap_or(1),
- )
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to create topic with name: {name} in stream ID: {stream_id}")
- })?;
- Ok((topic_id, partition_ids))
- }
-
- pub async fn update_topic_bypass_auth(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- name: &str,
- message_expiry: IggyExpiry,
- compression_algorithm: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- replication_factor: Option<u8>,
- ) -> Result<(), IggyError> {
- self.update_topic_base(
- stream_id,
- topic_id,
- name,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor,
- )
- .await
- }
-
pub fn update_topic2(
&self,
session: &Session,
@@ -375,35 +150,27 @@ impl IggyShard {
self.ensure_authenticated(session)?;
//self.ensure_topic_exists(stream_id, topic_id)?;
{
- let topic_id = self
- .streams2
- .with_topic_root_by_id(stream_id, topic_id, |topic| topic.id());
- let stream_id = self
+ let topic_id_val = self.streams2.with_topic_by_id(
+ stream_id,
+ topic_id,
+ topics::helpers::get_topic_id(),
+ );
+ let stream_id_val = self
.streams2
- .with_root_by_id(stream_id, |stream| stream.id());
+ .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
self.permissioner.borrow().update_topic(
session.get_user_id(),
- stream_id as u32,
- topic_id as u32
+ stream_id_val as u32,
+ topic_id_val as u32
).with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - permission denied to update topic for user with id: {}, stream ID: {}, topic ID: {}",
session.get_user_id(),
- stream_id,
- topic_id,
+ stream_id_val,
+ topic_id_val,
)
})?;
}
- let exists = self
- .streams2
- .with_topic_root_by_id(stream_id, topic_id, |topic| {
- let old_name = topic.name();
- old_name == &name
- });
- if exists {
- // TODO: Fix the errors to accept Identifier instead of u32.
- return Err(IggyError::TopicNameAlreadyExists(name, 0));
- }
self.update_topic_base2(
stream_id,
@@ -449,142 +216,20 @@ impl IggyShard {
max_topic_size: MaxTopicSize,
replication_factor: u8,
) {
- let (old_name, new_name) =
- self.streams2
- .with_topic_root_by_id_mut(stream_id, topic_id, |topic| {
- let old_name = topic.name().clone();
- topic.set_name(name.clone());
- topic.set_message_expiry(message_expiry);
- topic.set_compression(compression_algorithm);
- topic.set_max_topic_size(max_topic_size);
- topic.set_replication_factor(replication_factor);
- (old_name, name)
- // TODO: Set message expiry for all partitions and segments.
- });
- if old_name != new_name {
- self.streams2.with_topics(stream_id, |topics| {
- topics.with_index_mut(|index| {
- // Rename the key inside of hashmap
- let idx = index.remove(&old_name).expect("Rename key: key not found");
- index.insert(new_name, idx);
- })
- });
- }
- }
-
- #[allow(clippy::too_many_arguments)]
- pub async fn update_topic(
- &self,
- session: &Session,
- stream_id: &Identifier,
- topic_id: &Identifier,
- name: &str,
- message_expiry: IggyExpiry,
- compression_algorithm: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- replication_factor: Option<u8>,
- ) -> Result<(), IggyError> {
- self.ensure_authenticated(session)?;
- {
- let stream = self.get_stream(stream_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
- })?;
- let topic = self
- .find_topic(session, &stream, topic_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to find topic with ID: {topic_id}"
- )
- })?;
- self.permissioner.borrow().update_topic(
- session.get_user_id(),
- topic.stream_id,
- topic.topic_id,
- ).with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to update topic for user with id: {}, stream ID: {}, topic ID: {}",
- session.get_user_id(),
- topic.stream_id,
- topic.topic_id,
- )
- })?;
- }
-
- self.update_topic_base(
- stream_id,
- topic_id,
- name,
+ let update_topic_closure = topics::helpers::update_topic(
+ name.clone(),
message_expiry,
compression_algorithm,
max_topic_size,
replication_factor,
- )
- .await
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to update topic with ID: {topic_id} in stream with ID: {stream_id}",
- )
- })?;
-
- // TODO: if message_expiry is changed, we need to check if we need to purge messages based on the new expiry
- // TODO: if max_size_bytes is changed, we need to check if we need to purge messages based on the new size
- // TODO: if replication_factor is changed, we need to do `something`
- Ok(())
- }
-
- async fn update_topic_base(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- name: &str,
- message_expiry: IggyExpiry,
- compression_algorithm: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- replication_factor: Option<u8>,
- ) -> Result<(), IggyError> {
- let new_name = self.get_stream_mut(stream_id)?
- .update_topic(
- topic_id,
- name,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor.unwrap_or(1),
- )
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to update topic with ID: {topic_id} in stream with ID: {stream_id}",
- )
- })?;
- let stream = self.get_stream(stream_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
- })?;
- let topic = stream
- .get_topic(&Identifier::from_str(new_name.as_str()).unwrap())
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id} in stream with ID: {stream_id}",
- )
- })?;
- for partition in topic.partitions.values() {
- let mut partition = partition.write().await;
- partition.message_expiry = message_expiry;
- for segment in partition.segments.iter_mut() {
- segment.update_message_expiry(message_expiry);
- }
+ );
+ let (old_name, new_name) =
+ self.streams2
+ .with_topic_by_id_mut(stream_id, topic_id, update_topic_closure);
+ if old_name != new_name {
+ let rename_closure = topics::helpers::rename_index(&old_name, new_name);
+ self.streams2.with_topics(stream_id, rename_closure);
}
-
- info!("Updated topic: {topic}");
- Ok(())
- }
-
- pub async fn delete_topic_bypass_auth(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- ) -> Result<Topic, IggyError> {
- let topic = self.delete_topic_base(stream_id, topic_id).await?;
- Ok(topic)
}
pub async fn delete_topic2(
@@ -596,12 +241,14 @@ impl IggyShard {
self.ensure_authenticated(session)?;
self.ensure_topic_exists(stream_id, topic_id)?;
{
- let topic_id = self
- .streams2
- .with_topic_root_by_id(stream_id, topic_id, |topic| topic.id());
+ let topic_id = self.streams2.with_topic_by_id(
+ stream_id,
+ topic_id,
+ topics::helpers::get_topic_id(),
+ );
let stream_id = self
.streams2
- .with_root_by_id(stream_id, |stream| stream.id());
+ .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
self.permissioner
.borrow()
.delete_topic(session.get_user_id(), stream_id as u32, topic_id as u32)
@@ -633,74 +280,11 @@ impl IggyShard {
stream_id: &Identifier,
topic_id: &Identifier,
) -> Result<topic2::Topic, IggyError> {
- let topic = self.streams2.with_root_by_id_mut(stream_id, |stream| {
- let topics = stream.topics();
- let id = topics.with_root_by_id(topic_id, |topic| topic.id());
- let topic = topics.delete(id);
- assert_eq!(topic.id(), id, "delete_topic: topic ID mismatch");
- topic
- });
+ let delete_topic_closure = topics::helpers::delete_topic(topic_id);
+ let topic = self.streams2.with_topics(stream_id, delete_topic_closure);
Ok(topic)
}
- pub async fn delete_topic(
- &self,
- session: &Session,
- stream_id: &Identifier,
- topic_id: &Identifier,
- ) -> Result<Topic, IggyError> {
- self.ensure_authenticated(session)?;
- {
- let stream = self.get_stream(stream_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
- })?;
- let topic = self
- .find_topic(session, &stream, topic_id)
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to find topic with ID: {topic_id} in stream with ID: {stream_id}")
- })?;
- self.permissioner.borrow().delete_topic(
- session.get_user_id(),
- topic.stream_id,
- topic.topic_id,
- ).with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to delete topic with ID: {topic_id} in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
- )
- })?;
- }
-
- let topic = self.delete_topic_base(stream_id, topic_id)
- .await
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to delete topic with ID: {topic_id} in stream with ID: {stream_id}")
- })?;
- Ok(topic)
- }
-
- async fn delete_topic_base(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- ) -> Result<Topic, IggyError> {
- let mut stream = self.get_stream_mut(stream_id)?;
- let stream_id_value = stream.stream_id;
- let topic = stream
- .delete_topic(topic_id)
- .with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to delete topic with ID: {topic_id} in stream with ID: {stream_id}"))?;
- drop(stream);
- self.metrics.decrement_topics(1);
- self.metrics
- .decrement_partitions(topic.get_partitions_count());
- self.metrics.decrement_messages(topic.get_messages_count());
- self.metrics
- .decrement_segments(topic.get_segments_count().await);
- self.client_manager
- .borrow_mut()
- .delete_consumer_groups_for_topic(stream_id_value, topic.topic_id);
- Ok(topic)
- }
pub async fn purge_topic2(
&self,
session: &Session,
@@ -709,12 +293,14 @@ impl IggyShard {
) -> Result<(), IggyError> {
self.ensure_authenticated(session)?;
{
- let topic_id = self
- .streams2
- .with_topic_root_by_id(stream_id, topic_id, |topic| topic.id());
+ let topic_id = self.streams2.with_topic_by_id(
+ stream_id,
+ topic_id,
+ topics::helpers::get_topic_id(),
+ );
let stream_id = self
.streams2
- .with_root_by_id(stream_id, |stream| stream.id());
+ .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
self.permissioner.borrow().purge_topic(
session.get_user_id(),
stream_id as u32,
@@ -732,75 +318,16 @@ impl IggyShard {
async fn purge_topic_base2(
&self,
- stream_id: &Identifier,
- topic_id: &Identifier,
+ _stream_id: &Identifier,
+ _topic_id: &Identifier,
) -> Result<(), IggyError> {
+ /*
self.streams2
.with_partitions(stream_id, topic_id, |partitions| {
//partitions
});
+ */
Ok(())
}
-
- pub async fn purge_topic(
- &self,
- session: &Session,
- stream_id: &Identifier,
- topic_id: &Identifier,
- ) -> Result<(), IggyError> {
- let stream = self.get_stream(stream_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
- })?;
- let topic = self
- .find_topic(session, &stream, topic_id)
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to find topic with ID: {topic_id} in stream with ID: {stream_id}")
- })?;
- self.permissioner
- .borrow()
- .purge_topic(session.get_user_id(), topic.stream_id, topic.topic_id)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - permission denied to purge topic with ID: {topic_id} in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
- )
- })?;
-
- self.purge_topic_base(topic.stream_id, topic.topic_id).await
- }
-
- pub async fn purge_topic_bypass_auth(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- ) -> Result<(), IggyError> {
- let stream = self.get_stream(stream_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
- })?;
- let topic = stream
- .get_topic(topic_id)
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id} in stream with ID: {stream_id}")
- })?;
-
- self.purge_topic_base(stream.stream_id, topic.topic_id)
- .await
- }
-
- async fn purge_topic_base(&self, stream_id: u32, topic_id: u32) -> Result<(), IggyError> {
- let stream = self.get_stream(&Identifier::numeric(stream_id)?)?;
- let topic = stream.get_topic(&Identifier::numeric(topic_id)?)?;
-
- for partition in topic.get_partitions() {
- let mut partition = partition.write().await;
- let partition_id = partition.partition_id;
- let namespace = IggyNamespace::new(stream_id, topic_id, partition_id);
- let shard_info = self.find_shard_table_record(&namespace).unwrap();
- if shard_info.id() == self.id {
- partition.purge().await?;
- }
- }
- Ok(())
- }
}
diff --git a/core/server/src/shard/system/utils.rs b/core/server/src/shard/system/utils.rs
index 28db67dc..ad23d984 100644
--- a/core/server/src/shard/system/utils.rs
+++ b/core/server/src/shard/system/utils.rs
@@ -1,6 +1,6 @@
use iggy_common::{Identifier, IggyError};
-use crate::{shard::IggyShard, slab::traits_ext::EntityComponentSystem};
+use crate::shard::IggyShard;
impl IggyShard {
pub fn ensure_stream_exists(&self, stream_id: &Identifier) -> Result<(), IggyError> {
@@ -15,11 +15,10 @@ impl IggyShard {
stream_id: &Identifier,
topic_id: &Identifier,
) -> Result<(), IggyError> {
- //self.ensure_stream_exists(stream_id)?;
- let stream_id = self.streams2.get_index(stream_id);
+ self.ensure_stream_exists(stream_id)?;
let exists = self
.streams2
- .with_by_id(stream_id, |(root, _)| root.topics().exists(topic_id));
+ .with_topics(stream_id, |topics| topics.exists(topic_id));
if !exists {
return Err(IggyError::TopicIdNotFound(0, 0));
}
@@ -36,7 +35,7 @@ impl IggyShard {
//self.ensure_topic_exists(stream_id, topic_id)?;
let exists = self
.streams2
- .with_topic_root_by_id(stream_id, topic_id, |root| {
+ .with_topic_by_id(stream_id, topic_id, |(root, _)| {
root.consumer_groups().exists(group_id)
});
if !exists {
diff --git a/core/server/src/slab/helpers.rs b/core/server/src/slab/helpers.rs
new file mode 100644
index 00000000..20d36014
--- /dev/null
+++ b/core/server/src/slab/helpers.rs
@@ -0,0 +1,50 @@
+use crate::{
+ slab::{partitions::Partitions, topics::Topics, traits_ext::ComponentsById},
+ streaming::{
+ streams::stream2::{StreamRef, StreamRefMut},
+ topics::topic2::{TopicRef, TopicRefMut},
+ },
+};
+
+// Helpers
+pub fn topics<O, F>(f: F) -> impl FnOnce(ComponentsById<StreamRef>) -> O
+where
+ F: for<'a> FnOnce(&'a Topics) -> O,
+{
+ |(root, _)| f(root.topics())
+}
+
+pub async fn topics_async<O, F>(f: F) -> impl AsyncFnOnce(ComponentsById<StreamRef>) -> O
+where
+ F: for<'a> AsyncFnOnce(&'a Topics) -> O,
+{
+ async |(root, _)| f(root.topics()).await
+}
+
+pub fn topics_mut<O, F>(f: F) -> impl FnOnce(ComponentsById<StreamRefMut>) -> O
+where
+ F: for<'a> FnOnce(&'a mut Topics) -> O,
+{
+ |(mut root, _)| f(root.topics_mut())
+}
+
+pub fn partitions<O, F>(f: F) -> impl FnOnce(ComponentsById<TopicRef>) -> O
+where
+ F: for<'a> FnOnce(&'a Partitions) -> O,
+{
+ |(root, _)| f(root.partitions())
+}
+
+pub async fn partitions_async<O, F>(f: F) -> impl AsyncFnOnce(ComponentsById<TopicRef>) -> O
+where
+ F: for<'a> AsyncFnOnce(&'a Partitions) -> O,
+{
+ async |(root, _)| f(root.partitions()).await
+}
+
+pub fn partitions_mut<O, F>(f: F) -> impl FnOnce(ComponentsById<TopicRefMut>) -> O
+where
+ F: for<'a> FnOnce(&'a mut Partitions) -> O,
+{
+ |(mut root, _)| f(root.partitions_mut())
+}
diff --git a/core/server/src/slab/mod.rs b/core/server/src/slab/mod.rs
index 840de3d3..1ec4dfd0 100644
--- a/core/server/src/slab/mod.rs
+++ b/core/server/src/slab/mod.rs
@@ -1,19 +1,24 @@
pub mod consumer_groups;
+/// General rules how to implement `with_*` methods on any slab"
+/// 1. When implementing method that accepts closure f, make sure that the caller can supply closure with only 1 depth.
+/// for example, observe following code snippet:
+/// ```rust
+/// let topic_id = self.streams2.with_topic_by_id(stream_id, topic_id, get_topic_id());
+/// ```
+/// if we would not provide a `with_topic_by_id` method and purely relied only ony `with_topics`, we would have to write:
+/// ```rust
+/// let topic_id = self.streams2.with_topics(stream_id, get_topic_by_id(topic_id, get_topic_id())); // `get_topic_id` is a closure that retrieves the topic id.
+/// ```
+/// we need to supply a nested closure to `get_topic_by_id`.
+pub mod helpers;
pub mod partitions;
pub mod streams;
pub mod topics;
pub mod traits_ext;
-use ahash::AHashMap;
-use slab::Slab;
-use std::{
- fmt::{Debug, Display},
- ops::{Index, IndexMut},
-};
+use std::fmt::{Debug, Display};
pub trait Keyed {
type Key: Eq + std::hash::Hash + Clone + Debug + Display;
fn key(&self) -> &Self::Key;
}
-
-//index: AHashMap<T::Key, usize>,
diff --git a/core/server/src/slab/partitions.rs b/core/server/src/slab/partitions.rs
index 63f174d0..eed18c12 100644
--- a/core/server/src/slab/partitions.rs
+++ b/core/server/src/slab/partitions.rs
@@ -100,18 +100,18 @@ impl<'a> From<&'a Partitions> for PartitionRef<'a> {
impl EntityComponentSystem<Borrow> for Partitions {
type Idx = ContainerId;
type Entity = Partition;
- type EntityRef<'a> = PartitionRef<'a>;
+ type EntityComponents<'a> = PartitionRef<'a>;
- fn with<O, F>(&self, f: F) -> O
+ fn with_components<O, F>(&self, f: F) -> O
where
- F: for<'a> FnOnce(Self::EntityRef<'a>) -> O,
+ F: for<'a> FnOnce(Self::EntityComponents<'a>) -> O,
{
f(self.into())
}
- async fn with_async<O, F>(&self, f: F) -> O
+ async fn with_components_async<O, F>(&self, f: F) -> O
where
- F: for<'a> AsyncFnOnce(Self::EntityRef<'a>) -> O,
+ F: for<'a> AsyncFnOnce(Self::EntityComponents<'a>) -> O,
{
f(self.into()).await
}
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 15ed08ac..7f93667a 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -1,22 +1,24 @@
-use ahash::AHashMap;
-use iggy_common::Identifier;
-use slab::Slab;
-use std::{cell::RefCell, sync::Arc};
-use tokio_rustls::StartHandshake;
-
use crate::{
- shard::stats,
slab::{
- Keyed,
+ Keyed, helpers,
partitions::Partitions,
topics::Topics,
traits_ext::{
- Delete, DeleteCell, EntityComponentSystem, EntityComponentSystemMutCell, Insert,
- InsertCell, InteriorMutability, IntoComponents,
+ Components, ComponentsById, ComponentsByIdMapping, ComponentsMapping, DeleteCell,
+ EntityComponentSystem, EntityComponentSystemMutCell, InsertCell, InteriorMutability,
+ IntoComponents, IntoComponentsById,
},
},
- streaming::{stats::stats::StreamStats, streams::stream2, topics::topic2},
+ streaming::{
+ stats::stats::StreamStats,
+ streams::stream2::{self, StreamRef, StreamRefMut},
+ topics::topic2::{self, TopicRef, TopicRefMut},
+ },
};
+use ahash::AHashMap;
+use iggy_common::Identifier;
+use slab::Slab;
+use std::{cell::RefCell, sync::Arc};
const CAPACITY: usize = 1024;
pub type ContainerId = usize;
@@ -74,29 +76,29 @@ impl DeleteCell for Streams {
impl EntityComponentSystem<InteriorMutability> for Streams {
type Idx = ContainerId;
type Entity = stream2::Stream;
- type EntityRef<'a> = stream2::StreamRef<'a>;
+ type EntityComponents<'a> = stream2::StreamRef<'a>;
- fn with<O, F>(&self, f: F) -> O
+ fn with_components<O, F>(&self, f: F) -> O
where
- F: for<'a> FnOnce(Self::EntityRef<'a>) -> O,
+ F: for<'a> FnOnce(Self::EntityComponents<'a>) -> O,
{
f(self.into())
}
- async fn with_async<O, F>(&self, f: F) -> O
+ async fn with_components_async<O, F>(&self, f: F) -> O
where
- F: for<'a> AsyncFnOnce(Self::EntityRef<'a>) -> O,
+ F: for<'a> AsyncFnOnce(Self::EntityComponents<'a>) -> O,
{
f(self.into()).await
}
}
impl EntityComponentSystemMutCell for Streams {
- type EntityRefMut<'a> = stream2::StreamRefMut<'a>;
+ type EntityComponentsMut<'a> = stream2::StreamRefMut<'a>;
- fn with_mut<O, F>(&self, f: F) -> O
+ fn with_components_mut<O, F>(&self, f: F) -> O
where
- F: for<'a> FnOnce(Self::EntityRefMut<'a>) -> O,
+ F: for<'a> FnOnce(Self::EntityComponentsMut<'a>) -> O,
{
f(self.into())
}
@@ -134,79 +136,52 @@ impl Streams {
}
}
- pub fn with_root_by_id<T>(
+ pub fn with_index<T>(
&self,
- id: &Identifier,
- f: impl FnOnce(&stream2::StreamRoot) -> T,
+ f: impl FnOnce(&AHashMap<<stream2::StreamRoot as Keyed>::Key, usize>) -> T,
) -> T {
- let id = self.get_index(id);
- self.with_by_id(id, |(root, _)| f(&root))
+ let index = self.index.borrow();
+ f(&index)
}
- pub async fn with_root_by_id_async<T>(
+ pub fn with_index_mut<T>(
&self,
- id: &Identifier,
- f: impl AsyncFnOnce(&stream2::StreamRoot) -> T,
+ f: impl FnOnce(&mut AHashMap<<stream2::StreamRoot as Keyed>::Key, usize>) -> T,
) -> T {
- let id = self.get_index(id);
- self.with_by_id_async(id, async |(root, _)| f(&root).await)
- .await
+ let mut index = self.index.borrow_mut();
+ f(&mut index)
}
- pub fn with_root_by_id_mut<T>(
+ pub fn with_stream_by_id<T>(
&self,
id: &Identifier,
- f: impl FnOnce(&mut stream2::StreamRoot) -> T,
+ f: impl FnOnce(ComponentsById<StreamRef>) -> T,
) -> T {
let id = self.get_index(id);
- self.with_by_id_mut(id, |(mut root, _)| f(&mut root))
- }
-
- pub fn with_stats<T>(&self, f: impl FnOnce(&Slab<Arc<StreamStats>>) -> T) -> T {
- self.with(|components| {
- let (_, stats) = components.into_components();
- f(&stats)
- })
+ self.with_components_by_id(id, |stream| f(stream))
}
- pub fn with_stats_by_id<T>(
+ pub async fn with_stream_by_id_async<T>(
&self,
id: &Identifier,
- f: impl FnOnce(&Arc<StreamStats>) -> T,
+ f: impl AsyncFnOnce(ComponentsById<StreamRef>) -> T,
) -> T {
let id = self.get_index(id);
- self.with_by_id(id, |(_, stats)| {
- let stats = stats;
- f(&stats)
- })
- }
-
- pub fn with_stats_mut<T>(&self, f: impl FnOnce(&mut Slab<Arc<StreamStats>>) -> T) -> T {
- self.with_mut(|components| {
- let (_, mut stats) = components.into_components();
- f(&mut stats)
- })
- }
-
- pub fn with_index<T>(
- &self,
- f: impl FnOnce(&AHashMap<<stream2::StreamRoot as Keyed>::Key, usize>) -> T,
- ) -> T {
- let index = self.index.borrow();
- f(&index)
+ self.with_components_by_id_async(id, async |stream| f(stream).await)
+ .await
}
- pub fn with_index_mut<T>(
+ pub fn with_stream_by_id_mut<T>(
&self,
- f: impl FnOnce(&mut AHashMap<<stream2::StreamRoot as Keyed>::Key, usize>) -> T,
+ id: &Identifier,
+ f: impl FnOnce(ComponentsById<StreamRefMut>) -> T,
) -> T {
- let mut index = self.index.borrow_mut();
- f(&mut index)
+ let id = self.get_index(id);
+ self.with_components_by_id_mut(id, |stream| f(stream))
}
pub fn with_topics<T>(&self, stream_id: &Identifier, f: impl FnOnce(&Topics) -> T) -> T {
- let id = self.get_index(stream_id);
- self.with_by_id(id, |(root, _)| f(root.topics()))
+ self.with_stream_by_id(stream_id, helpers::topics(f))
}
pub async fn with_topics_async<T>(
@@ -214,8 +189,7 @@ impl Streams {
stream_id: &Identifier,
f: impl AsyncFnOnce(&Topics) -> T,
) -> T {
- let id = self.get_index(stream_id);
- self.with_by_id_async(id, async |(root, _)| f(root.topics()).await)
+ self.with_stream_by_id_async(stream_id, helpers::topics_async(f).await)
.await
}
@@ -224,80 +198,75 @@ impl Streams {
stream_id: &Identifier,
f: impl FnOnce(&mut Topics) -> T,
) -> T {
- let id = self.get_index(stream_id);
- self.with_by_id_mut(id, |(mut root, _)| {
- let topics = root.topics_mut();
- f(topics)
- })
+ self.with_stream_by_id_mut(stream_id, helpers::topics_mut(f))
}
- pub fn with_topic_root_by_id<T>(
+ pub fn with_topic_by_id<T>(
&self,
stream_id: &Identifier,
- id: &Identifier,
- f: impl FnOnce(&topic2::TopicRoot) -> T,
+ topic_id: &Identifier,
+ f: impl FnOnce(ComponentsById<TopicRef>) -> T,
) -> T {
- self.with_topics(stream_id, |topics| {
- topics.with_root_by_id(id, |root| f(root))
+ self.with_topics(stream_id, |container| {
+ container.with_topic_by_id(topic_id, f)
})
}
- pub async fn with_topic_root_by_id_async<T>(
+ pub async fn with_topic_by_id_async<T>(
&self,
stream_id: &Identifier,
- id: &Identifier,
- f: impl AsyncFnOnce(&topic2::TopicRoot) -> T,
+ topic_id: &Identifier,
+ f: impl AsyncFnOnce(ComponentsById<TopicRef>) -> T,
) -> T {
- self.with_topics_async(stream_id, async |topics| {
- let id = topics.get_index(id);
- topics
- .with_by_id_async(id, async |(root, _)| f(&root).await)
- .await
+ self.with_topics_async(stream_id, async |container| {
+ container.with_topic_by_id_async(topic_id, f).await
})
.await
}
- pub fn with_topic_root_by_id_mut<T>(
+ pub fn with_topic_by_id_mut<T>(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
- f: impl FnOnce(&mut topic2::TopicRoot) -> T,
+ f: impl FnOnce(ComponentsById<TopicRefMut>) -> T,
) -> T {
- self.with_topics_mut(stream_id, |topics| {
- topics.with_root_by_id_mut(topic_id, |root| f(root))
+ self.with_topics_mut(stream_id, |container| {
+ container.with_topic_by_id_mut(topic_id, f)
})
}
- pub fn with_partitions(
+ pub fn with_partitions<T>(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
- f: impl FnOnce(&Partitions),
- ) {
- self.with_topics(stream_id, |topics| {
- topics.with_partitions(topic_id, f);
- });
+ f: impl FnOnce(&Partitions) -> T,
+ ) -> T {
+ self.with_topics(stream_id, |container| {
+ container.with_partitions(topic_id, f)
+ })
}
- pub fn with_partitions_mut<T>(
+ pub async fn with_partitions_async<T>(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
- f: impl FnOnce(&mut Partitions) -> T,
+ f: impl AsyncFnOnce(&Partitions) -> T,
) -> T {
- self.with_topics_mut(stream_id, |topics| topics.with_partitions_mut(topic_id, f))
+ self.with_topics_async(stream_id, async |container| {
+ container.with_partitions_async(topic_id, f).await
+ })
+ .await
}
- pub async fn with_partitions_async<T>(
+ pub fn with_partitions_mut<T>(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
- f: impl AsyncFnOnce(&Partitions) -> T,
+ f: impl FnOnce(&mut Partitions) -> T,
) -> T {
- self.with_topics_async(stream_id, async |topics| {
- topics.with_partitions_async(topic_id, f).await
+ self.with_topics_mut(stream_id, |container| {
+ container.with_partitions_mut(topic_id, f)
})
- .await
}
pub fn len(&self) -> usize {
diff --git a/core/server/src/slab/topics.rs b/core/server/src/slab/topics.rs
index 87eb360d..eb5af5ba 100644
--- a/core/server/src/slab/topics.rs
+++ b/core/server/src/slab/topics.rs
@@ -5,17 +5,17 @@ use std::{cell::RefCell, sync::Arc};
use crate::{
slab::{
- Keyed,
+ Keyed, helpers,
partitions::Partitions,
traits_ext::{
- Delete, DeleteCell, EntityComponentSystem, EntityComponentSystemMutCell, Insert,
- InsertCell, InteriorMutability, IntoComponents,
+ ComponentsById, Delete, DeleteCell, EntityComponentSystem,
+ EntityComponentSystemMutCell, Insert, InsertCell, InteriorMutability, IntoComponents,
},
},
streaming::{
partitions::partition2,
stats::stats::TopicStats,
- topics::topic2::{self, TopicRef},
+ topics::topic2::{self, TopicRef, TopicRefMut},
},
};
@@ -86,29 +86,29 @@ impl<'a> From<&'a Topics> for topic2::TopicRefMut<'a> {
impl EntityComponentSystem<InteriorMutability> for Topics {
type Idx = ContainerId;
type Entity = topic2::Topic;
- type EntityRef<'a> = topic2::TopicRef<'a>;
+ type EntityComponents<'a> = topic2::TopicRef<'a>;
- fn with<O, F>(&self, f: F) -> O
+ fn with_components<O, F>(&self, f: F) -> O
where
- F: for<'a> FnOnce(Self::EntityRef<'a>) -> O,
+ F: for<'a> FnOnce(Self::EntityComponents<'a>) -> O,
{
f(self.into())
}
- async fn with_async<O, F>(&self, f: F) -> O
+ async fn with_components_async<O, F>(&self, f: F) -> O
where
- F: for<'a> AsyncFnOnce(Self::EntityRef<'a>) -> O,
+ F: for<'a> AsyncFnOnce(Self::EntityComponents<'a>) -> O,
{
f(self.into()).await
}
}
impl EntityComponentSystemMutCell for Topics {
- type EntityRefMut<'a> = topic2::TopicRefMut<'a>;
+ type EntityComponentsMut<'a> = topic2::TopicRefMut<'a>;
- fn with_mut<O, F>(&self, f: F) -> O
+ fn with_components_mut<O, F>(&self, f: F) -> O
where
- F: for<'a> FnOnce(Self::EntityRefMut<'a>) -> O,
+ F: for<'a> FnOnce(Self::EntityComponentsMut<'a>) -> O,
{
f(self.into())
}
@@ -158,51 +158,36 @@ impl Topics {
f(&mut index)
}
- pub fn with_root_by_id<T>(
+ pub fn with_topic_by_id<T>(
&self,
- id: &Identifier,
- f: impl FnOnce(&topic2::TopicRoot) -> T,
+ topic_id: &Identifier,
+ f: impl FnOnce(ComponentsById<TopicRef>) -> T,
) -> T {
- let id = self.get_index(id);
- self.with_by_id(id, |(root, _)| f(&root))
+ let id = self.get_index(topic_id);
+ self.with_components_by_id(id, |components| f(components))
}
- pub async fn with_root_by_id_async<T>(
+ pub async fn with_topic_by_id_async<T>(
&self,
- id: &Identifier,
- f: impl AsyncFnOnce(&topic2::TopicRoot) -> T,
+ topic_id: &Identifier,
+ f: impl AsyncFnOnce(ComponentsById<TopicRef>) -> T,
) -> T {
- let id = self.get_index(id);
- self.with_by_id_async(id, async |(root, _)| f(&root).await)
+ let id = self.get_index(topic_id);
+ self.with_components_by_id_async(id, async |components| f(components).await)
.await
}
- pub fn with_root_by_id_mut<T>(
+ pub fn with_topic_by_id_mut<T>(
&self,
- id: &Identifier,
- f: impl FnOnce(&mut topic2::TopicRoot) -> T,
+ topic_id: &Identifier,
+ f: impl FnOnce(ComponentsById<TopicRefMut>) -> T,
) -> T {
- let id = self.get_index(id);
- self.with_by_id_mut(id, |(mut root, _)| f(&mut root))
- }
-
- pub fn with_stats<T>(&self, f: impl FnOnce(&Slab<Arc<TopicStats>>) -> T) -> T {
- self.with(|components| {
- let (_, stats) = components.into_components();
- f(&stats)
- })
- }
-
- pub fn with_stats_mut<T>(&self, f: impl FnOnce(&mut Slab<Arc<TopicStats>>) -> T) -> T {
- self.with_mut(|components| {
- let (_, mut stats) = components.into_components();
- f(&mut stats)
- })
+ let id = self.get_index(topic_id);
+ self.with_components_by_id_mut(id, |components| f(components))
}
- pub fn with_stats_by_id<T>(&self, id: &Identifier, f: impl FnOnce(&Arc<TopicStats>) -> T) -> T {
- let id = self.get_index(id);
- self.with_by_id(id, |(_, stats)| f(&stats))
+ pub fn with_partitions<T>(&self, topic_id: &Identifier, f: impl FnOnce(&Partitions) -> T) -> T {
+ self.with_topic_by_id(topic_id, helpers::partitions(f))
}
pub async fn with_partitions_async<T>(
@@ -210,20 +195,8 @@ impl Topics {
topic_id: &Identifier,
f: impl AsyncFnOnce(&Partitions) -> T,
) -> T {
- let id = self.get_index(topic_id);
- self.with_by_id_async(id, async |(root, _)| {
- let partitions = root.partitions();
- f(partitions).await
- })
- .await
- }
-
- pub fn with_partitions(&self, topic_id: &Identifier, f: impl FnOnce(&Partitions)) {
- let id = self.get_index(topic_id);
- self.with_by_id(id, |(root, _)| {
- let partitions = root.partitions();
- f(partitions)
- })
+ self.with_topic_by_id_async(topic_id, helpers::partitions_async(f).await)
+ .await
}
pub fn with_partitions_mut<T>(
@@ -231,10 +204,6 @@ impl Topics {
topic_id: &Identifier,
f: impl FnOnce(&mut Partitions) -> T,
) -> T {
- let id = self.get_index(topic_id);
- self.with_by_id_mut(id, |(mut root, _)| {
- let partitions = root.partitions_mut();
- f(partitions)
- })
+ self.with_topic_by_id_mut(topic_id, helpers::partitions_mut(f))
}
}
diff --git a/core/server/src/slab/traits_ext.rs b/core/server/src/slab/traits_ext.rs
index 4b860f05..ac33243e 100644
--- a/core/server/src/slab/traits_ext.rs
+++ b/core/server/src/slab/traits_ext.rs
@@ -157,69 +157,78 @@ type MappingByIdMut<'a, E, T> =
pub type Components<T> = <T as IntoComponents>::Components;
pub type ComponentsById<'a, T> = <T as IntoComponentsById>::Output;
+// TODO: Not all hope is lost, don't get discouraged by the comment from above
+// I've figured there is actually and ergonomic improvement that can be made here.
+// Observe that the chain of constraints put on the `EntityRef` type is actually wrong.
+// We constraint the `EntityRef` to be IntoComponents + IntoComponentsById,
+// which from composability point of view is not ideal...
+// A better idea is to constraint the `IntoComponents::Output` of the `EntityRef` impl to - `IntoComponentsById`, rather than entire `EntityRef`.
+// This way we can name the type inside of the `with_by_id` methods, without desolving into the tuple mapping madness.
+// in the `stream.rs` `topic.rs` `partitions.rs` files, we need to implement the `IntoComponentsById` trait for the output type of `IntoComponents` implementation, for `EntityRef`.
+// to make our life easier, we can create a type alias for those tuples and maybe even create a macro, to not repeat the type 3 times per entity (TupleEntityType, TupleEntityTypeRef, TupleEntityTypeRefByid).
pub trait EntityComponentSystem<T>
where
<Self::Entity as IntoComponents>::Components: ComponentsMapping<T> + ComponentsByIdMapping<T>,
{
type Idx;
type Entity: IntoComponents + EntityMarker;
- type EntityRef<'a>: IntoComponents<Components = Mapping<'a, Self::Entity, T>>
+ type EntityComponents<'a>: IntoComponents<Components = Mapping<'a, Self::Entity, T>>
+ IntoComponentsById<Idx = Self::Idx, Output = MappingById<'a, Self::Entity, T>>;
- fn with<O, F>(&self, f: F) -> O
+ fn with_components<O, F>(&self, f: F) -> O
where
- F: for<'a> FnOnce(Self::EntityRef<'a>) -> O;
+ F: for<'a> FnOnce(Self::EntityComponents<'a>) -> O;
- fn with_async<O, F>(&self, f: F) -> impl Future<Output = O>
+ fn with_components_async<O, F>(&self, f: F) -> impl Future<Output = O>
where
- F: for<'a> AsyncFnOnce(Self::EntityRef<'a>) -> O;
+ F: for<'a> AsyncFnOnce(Self::EntityComponents<'a>) -> O;
- fn with_by_id<O, F>(&self, id: Self::Idx, f: F) -> O
+ fn with_components_by_id<O, F>(&self, id: Self::Idx, f: F) -> O
where
- F: for<'a> FnOnce(ComponentsById<'a, Self::EntityRef<'a>>) -> O,
+ F: for<'a> FnOnce(ComponentsById<'a, Self::EntityComponents<'a>>) -> O,
{
- self.with(|components| f(components.into_components_by_id(id)))
+ self.with_components(|components| f(components.into_components_by_id(id)))
}
- fn with_by_id_async<O, F>(&self, id: Self::Idx, f: F) -> impl Future<Output = O>
+ fn with_components_by_id_async<O, F>(&self, id: Self::Idx, f: F) -> impl Future<Output = O>
where
- F: for<'a> AsyncFnOnce(ComponentsById<'a, Self::EntityRef<'a>>) -> O,
+ F: for<'a> AsyncFnOnce(ComponentsById<'a, Self::EntityComponents<'a>>) -> O,
{
- self.with_async(async |components| f(components.into_components_by_id(id)).await)
+ self.with_components_async(async |components| f(components.into_components_by_id(id)).await)
}
}
pub trait EntityComponentSystemMut: EntityComponentSystem<Borrow> {
- type EntityRefMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, Borrow>>
+ type EntityComponentsMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, Borrow>>
+ IntoComponentsById<Idx = Self::Idx, Output = MappingByIdMut<'a, Self::Entity, Borrow>>;
- fn with_mut<O, F>(&mut self, f: F) -> O
+ fn with_components_mut<O, F>(&mut self, f: F) -> O
where
- F: for<'a> FnOnce(Self::EntityRefMut<'a>) -> O;
+ F: for<'a> FnOnce(Self::EntityComponentsMut<'a>) -> O;
- fn with_by_id_mut<O, F>(&mut self, id: Self::Idx, f: F) -> O
+ fn with_components_by_id_mut<O, F>(&mut self, id: Self::Idx, f: F) -> O
where
- F: for<'a> FnOnce(ComponentsById<'a, Self::EntityRefMut<'a>>) -> O,
+ F: for<'a> FnOnce(ComponentsById<'a, Self::EntityComponentsMut<'a>>) -> O,
{
- self.with_mut(|components| f(components.into_components_by_id(id)))
+ self.with_components_mut(|components| f(components.into_components_by_id(id)))
}
}
pub trait EntityComponentSystemMutCell: EntityComponentSystem<InteriorMutability> {
- type EntityRefMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, InteriorMutability>>
+ type EntityComponentsMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, InteriorMutability>>
+ IntoComponentsById<
Idx = Self::Idx,
Output = MappingByIdMut<'a, Self::Entity, InteriorMutability>,
>;
- fn with_mut<O, F>(&self, f: F) -> O
+ fn with_components_mut<O, F>(&self, f: F) -> O
where
- F: for<'a> FnOnce(Self::EntityRefMut<'a>) -> O;
+ F: for<'a> FnOnce(Self::EntityComponentsMut<'a>) -> O;
- fn with_by_id_mut<O, F>(&self, id: Self::Idx, f: F) -> O
+ fn with_components_by_id_mut<O, F>(&self, id: Self::Idx, f: F) -> O
where
- F: for<'a> FnOnce(ComponentsById<'a, Self::EntityRefMut<'a>>) -> O,
+ F: for<'a> FnOnce(ComponentsById<'a, Self::EntityComponentsMut<'a>>) -> O,
{
- self.with_mut(|components| f(components.into_components_by_id(id)))
+ self.with_components_mut(|components| f(components.into_components_by_id(id)))
}
}
diff --git a/core/server/src/streaming/partitions/partition2.rs b/core/server/src/streaming/partitions/partition2.rs
index c5633048..20af7651 100644
--- a/core/server/src/streaming/partitions/partition2.rs
+++ b/core/server/src/streaming/partitions/partition2.rs
@@ -113,6 +113,10 @@ impl PartitionRoot {
pub fn update_id(&mut self, id: usize) {
self.id = id;
}
+
+ pub fn created_at(&self) -> IggyTimestamp {
+ self.created_at
+ }
}
// TODO: Create a macro to impl those PartitionRef/PartitionRefMut structs and it's traits.
diff --git a/core/server/src/streaming/stats/stats.rs b/core/server/src/streaming/stats/stats.rs
index 7a834600..b5039c11 100644
--- a/core/server/src/streaming/stats/stats.rs
+++ b/core/server/src/streaming/stats/stats.rs
@@ -19,16 +19,16 @@ impl StreamStats {
}
}
- pub fn size_bytes(&self) -> u64 {
- self.size_bytes.load(Ordering::SeqCst)
+ pub fn size_bytes_inconsistent(&self) -> u64 {
+ self.size_bytes.load(Ordering::Relaxed)
}
- pub fn messages_count(&self) -> u64 {
- self.messages_count.load(Ordering::SeqCst)
+ pub fn messages_count_inconsistent(&self) -> u64 {
+ self.messages_count.load(Ordering::Relaxed)
}
- pub fn segments_count(&self) -> u32 {
- self.segments_count.load(Ordering::SeqCst)
+ pub fn segments_count_inconsistent(&self) -> u32 {
+ self.segments_count.load(Ordering::Relaxed)
}
}
@@ -47,6 +47,14 @@ impl TopicStats {
messages_count: AtomicU64::new(0),
}
}
+
+ pub fn size_bytes_inconsistent(&self) -> u64 {
+ self.size_bytes.load(Ordering::Relaxed)
+ }
+
+ pub fn messages_count_inconsistent(&self) -> u64 {
+ self.messages_count.load(Ordering::Relaxed)
+ }
}
#[derive(Default, Debug)]
@@ -54,6 +62,7 @@ pub struct PartitionStats {
parent: Arc<TopicStats>,
messages_count: AtomicU64,
size_bytes: AtomicU64,
+ segments_count: AtomicU32,
}
impl PartitionStats {
@@ -62,6 +71,19 @@ impl PartitionStats {
parent: parent_stats,
messages_count: AtomicU64::new(0),
size_bytes: AtomicU64::new(0),
+ segments_count: AtomicU32::new(0),
}
}
+
+ pub fn size_bytes_inconsistent(&self) -> u64 {
+ self.size_bytes.load(Ordering::Relaxed)
+ }
+
+ pub fn messages_count_inconsistent(&self) -> u64 {
+ self.messages_count.load(Ordering::Relaxed)
+ }
+
+ pub fn segments_count_inconsistent(&self) -> u32 {
+ self.segments_count.load(Ordering::Relaxed)
+ }
}
diff --git a/core/server/src/streaming/streams/helpers.rs b/core/server/src/streaming/streams/helpers.rs
new file mode 100644
index 00000000..be256f83
--- /dev/null
+++ b/core/server/src/streaming/streams/helpers.rs
@@ -0,0 +1,18 @@
+use crate::{
+ slab::{streams, traits_ext::ComponentsById},
+ streaming::streams::stream2::{StreamRef, StreamRefMut, StreamRoot},
+};
+
+pub fn get_stream_id() -> impl FnOnce(ComponentsById<StreamRef>) -> streams::ContainerId {
+ |(root, _)| root.id()
+}
+
+pub fn get_stream_name() -> impl FnOnce(ComponentsById<StreamRef>) -> String {
+ |(root, _)| root.name().clone()
+}
+
+pub fn update_stream_name(name: String) -> impl FnOnce(ComponentsById<StreamRefMut>) {
+ move |(mut root, _)| {
+ root.set_name(name);
+ }
+}
diff --git a/core/server/src/streaming/streams/mod.rs b/core/server/src/streaming/streams/mod.rs
index 8f78c7ff..6e6f6db2 100644
--- a/core/server/src/streaming/streams/mod.rs
+++ b/core/server/src/streaming/streams/mod.rs
@@ -16,6 +16,7 @@
* under the License.
*/
+pub mod helpers;
pub mod messages;
pub mod partitions;
pub mod persistence;
diff --git a/core/server/src/streaming/streams/stream2.rs b/core/server/src/streaming/streams/stream2.rs
index 309e4b95..1eca7a55 100644
--- a/core/server/src/streaming/streams/stream2.rs
+++ b/core/server/src/streaming/streams/stream2.rs
@@ -63,6 +63,10 @@ impl StreamRoot {
pub fn topics_mut(&mut self) -> &mut Topics {
&mut self.topics
}
+
+ pub fn created_at(&self) -> IggyTimestamp {
+ self.created_at
+ }
}
#[derive(Debug, Clone)]
diff --git a/core/server/src/streaming/streams/topics.rs b/core/server/src/streaming/streams/topics.rs
index cec40c33..098e576e 100644
--- a/core/server/src/streaming/streams/topics.rs
+++ b/core/server/src/streaming/streams/topics.rs
@@ -25,143 +25,14 @@ use iggy_common::CompressionAlgorithm;
use iggy_common::IggyError;
use iggy_common::IggyExpiry;
use iggy_common::MaxTopicSize;
-use iggy_common::locking::IggyRwLockFn;
use iggy_common::{IdKind, Identifier};
use std::sync::atomic::Ordering;
-use tracing::info;
impl Stream {
pub fn get_topics_count(&self) -> u32 {
self.topics.len() as u32
}
- #[allow(clippy::too_many_arguments)]
- pub fn create_topic(
- &mut self,
- topic_id: Option<u32>,
- name: &str,
- partitions_count: u32,
- message_expiry: IggyExpiry,
- compression_algorithm: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- replication_factor: u8,
- ) -> Result<(Identifier, Vec<u32>), IggyError> {
- let max_topic_size = Topic::get_max_topic_size(max_topic_size, &self.config)?;
- if self.topics_ids.contains_key(name) {
- return Err(IggyError::TopicNameAlreadyExists(
- name.to_owned(),
- self.stream_id,
- ));
- }
-
- let mut id;
- if topic_id.is_none() {
- id = self.current_topic_id.fetch_add(1, Ordering::SeqCst);
- loop {
- if self.topics.contains_key(&id) {
- if id == u32::MAX {
- return Err(IggyError::TopicIdAlreadyExists(id, self.stream_id));
- }
- id = self.current_topic_id.fetch_add(1, Ordering::SeqCst);
- } else {
- break;
- }
- }
- } else {
- id = topic_id.unwrap();
- }
-
- if self.topics.contains_key(&id) {
- return Err(IggyError::TopicIdAlreadyExists(id, self.stream_id));
- }
-
- let CreatedTopicInfo {
- topic,
- partition_ids,
- } = Topic::create(
- self.stream_id,
- id,
- name,
- partitions_count,
- self.config.clone(),
- self.storage.clone(),
- self.size_bytes.clone(),
- self.messages_count.clone(),
- self.segments_count.clone(),
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor,
- )?;
-
- self.topics_ids.insert(name.to_owned(), id);
- self.topics.insert(id, topic);
- Ok((Identifier::numeric(id).unwrap(), partition_ids))
- }
-
- pub fn update_topic(
- &mut self,
- id: &Identifier,
- name: &str,
- message_expiry: IggyExpiry,
- compression_algorithm: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- replication_factor: u8,
- ) -> Result<String, IggyError> {
- let message_expiry = Topic::get_message_expiry(message_expiry, &self.config);
- let max_topic_size = Topic::get_max_topic_size(max_topic_size, &self.config)?;
- let topic_id;
- {
- let topic = self.get_topic(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get topic with id: {id}")
- })?;
- topic_id = topic.topic_id;
- }
-
- {
- if let Some(topic_id_by_name) = self.topics_ids.get(name) {
- if *topic_id_by_name != topic_id {
- return Err(IggyError::TopicNameAlreadyExists(
- name.to_owned(),
- self.stream_id,
- ));
- }
- }
- }
-
- let old_topic_name = {
- let topic = self.get_topic(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get topic with id: {id}")
- })?;
- topic.name.clone()
- };
-
- {
- let topic = self.get_topic_mut(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get mutable reference to topic with id {id}")
- })?;
-
- topic.name = name.to_owned();
- topic.message_expiry = message_expiry;
- topic.compression_algorithm = compression_algorithm;
- topic.max_topic_size = max_topic_size;
- topic.replication_factor = replication_factor;
- }
-
- let temp = name.to_owned();
- self.topics_ids.remove(&old_topic_name.clone());
- self.topics_ids.insert(temp.clone(), topic_id);
-
- Ok(temp)
- }
-
- pub fn remove_topic(&mut self, identifier: &Identifier) -> Result<Topic, IggyError> {
- match identifier.kind {
- IdKind::Numeric => self.remove_topic_by_id(identifier.get_u32_value()?),
- IdKind::String => self.remove_topic_by_name(&identifier.get_cow_str_value()?),
- }
- }
-
pub fn get_topics(&self) -> Vec<&Topic> {
self.topics.values().collect()
}
@@ -241,80 +112,4 @@ impl Stream {
.remove(&topic_id)
.ok_or(IggyError::TopicIdNotFound(topic_id, self.stream_id))
}
-
- pub fn delete_topic(&mut self, id: &Identifier) -> Result<Topic, IggyError> {
- let topic = self.remove_topic(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to remove topic with id: {id}")
- })?;
- let topic_id = topic.topic_id;
- let current_topic_id = self.current_topic_id.load(Ordering::SeqCst);
- if current_topic_id > topic_id {
- self.current_topic_id.store(topic_id, Ordering::SeqCst);
- }
-
- Ok(topic)
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use crate::{
- configs::system::SystemConfig,
- streaming::{
- persistence::persister::{FileWithSyncPersister, PersisterKind},
- storage::SystemStorage,
- utils::MemoryPool,
- },
- };
- use iggy_common::IggyByteSize;
- use std::{rc::Rc, sync::Arc};
-
- #[tokio::test]
- async fn should_get_topic_by_id_and_name() {
- let tempdir = tempfile::TempDir::new().unwrap();
- let config = Arc::new(SystemConfig {
- path: tempdir.path().to_str().unwrap().to_string(),
- ..Default::default()
- });
- let storage = Rc::new(SystemStorage::new(
- config.clone(),
- Arc::new(PersisterKind::FileWithSync(FileWithSyncPersister {})),
- ));
- MemoryPool::init_pool(config.clone());
- let stream_id = 1;
- let stream_name = "test_stream";
- let topic_id = 2;
- let topic_name = "test_topic";
- let message_expiry = IggyExpiry::NeverExpire;
- let compression_algorithm = CompressionAlgorithm::None;
- let max_topic_size = 2 * config.segment.size.as_bytes_u64();
- let max_topic_size = MaxTopicSize::Custom(IggyByteSize::from(max_topic_size));
- let mut stream = Stream::create(stream_id, stream_name, config, storage);
- stream
- .create_topic(
- Some(topic_id),
- topic_name,
- 1,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- 1,
- )
- .unwrap();
-
- let topic = stream.get_topic(&Identifier::numeric(topic_id).unwrap());
- assert!(topic.is_ok());
- let topic = topic.unwrap();
- assert_eq!(topic.topic_id, topic_id);
- assert_eq!(topic.name, topic_name);
- assert_eq!(topic.compression_algorithm, compression_algorithm);
-
- let topic = stream.get_topic(&Identifier::named(topic_name).unwrap());
- assert!(topic.is_ok());
- let topic = topic.unwrap();
- assert_eq!(topic.topic_id, topic_id);
- assert_eq!(topic.name, topic_name);
- assert_eq!(topic.compression_algorithm, compression_algorithm);
- }
}
diff --git a/core/server/src/streaming/topics/helpers.rs b/core/server/src/streaming/topics/helpers.rs
new file mode 100644
index 00000000..6986c493
--- /dev/null
+++ b/core/server/src/streaming/topics/helpers.rs
@@ -0,0 +1,58 @@
+use crate::{
+ slab::{
+ Keyed,
+ topics::{self, Topics},
+ traits_ext::{ComponentsById, DeleteCell, EntityMarker},
+ },
+ streaming::topics::topic2::{Topic, TopicRef, TopicRefMut, TopicRoot},
+};
+use iggy_common::{CompressionAlgorithm, Identifier, IggyExpiry, MaxTopicSize};
+
+pub fn rename_index(
+ old_name: &<TopicRoot as Keyed>::Key,
+ new_name: String,
+) -> impl FnOnce(&Topics) {
+ move |topics| {
+ topics.with_index_mut(|index| {
+ // Rename the key inside of hashmap
+ let idx = index.remove(old_name).expect("Rename key: key not found");
+ index.insert(new_name, idx);
+ })
+ }
+}
+
+pub fn get_topic_id() -> impl FnOnce(ComponentsById<TopicRef>) -> topics::ContainerId {
+ |(root, _)| root.id()
+}
+
+pub fn delete_topic(topic_id: &Identifier) -> impl FnOnce(&Topics) -> Topic {
+ move |topics| {
+ let id = topics.get_index(topic_id);
+ let topic = topics.delete(id);
+ assert_eq!(topic.id(), id, "delete_topic: topic ID mismatch");
+ topic
+ }
+}
+
+pub fn exists(identifier: &Identifier) -> impl FnOnce(&Topics) -> bool {
+ move |topics| topics.exists(identifier)
+}
+
+pub fn update_topic(
+ name: String,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ replication_factor: u8,
+) -> impl FnOnce(ComponentsById<TopicRefMut>) -> (String, String) {
+ move |(mut root, _)| {
+ let old_name = root.name().clone();
+ root.set_name(name.clone());
+ root.set_message_expiry(message_expiry);
+ root.set_compression(compression_algorithm);
+ root.set_max_topic_size(max_topic_size);
+ root.set_replication_factor(replication_factor);
+ (old_name, name)
+ // TODO: Set message expiry for all partitions and segments.
+ }
+}
diff --git a/core/server/src/streaming/topics/mod.rs b/core/server/src/streaming/topics/mod.rs
index 0e7bea60..cab1f3bf 100644
--- a/core/server/src/streaming/topics/mod.rs
+++ b/core/server/src/streaming/topics/mod.rs
@@ -20,6 +20,7 @@ pub mod consumer_group;
pub mod consumer_group2;
pub mod consumer_groups;
pub mod consumer_offsets;
+pub mod helpers;
pub mod messages;
pub mod partitions;
pub mod persistence;
diff --git a/core/server/src/streaming/topics/storage.rs b/core/server/src/streaming/topics/storage.rs
index d4b3b805..fe726194 100644
--- a/core/server/src/streaming/topics/storage.rs
+++ b/core/server/src/streaming/topics/storage.rs
@@ -49,187 +49,7 @@ struct ConsumerGroupData {
impl TopicStorage for FileTopicStorage {
async fn load(&self, topic: &mut Topic, mut state: TopicState) -> Result<(), IggyError> {
- info!("Loading topic {} from disk...", topic);
- if !Path::new(&topic.path).exists() {
- return Err(IggyError::TopicIdNotFound(topic.topic_id, topic.stream_id));
- }
-
- let message_expiry = Topic::get_message_expiry(state.message_expiry, &topic.config);
- let max_topic_size = Topic::get_max_topic_size(state.max_topic_size, &topic.config)?;
- topic.created_at = state.created_at;
- topic.message_expiry = message_expiry;
- topic.max_topic_size = max_topic_size;
- topic.compression_algorithm = state.compression_algorithm;
- topic.replication_factor = state.replication_factor.unwrap_or(1);
-
- let mut dir_entries = std::fs::read_dir(&topic.partitions_path)
- .with_context(|| format!("Failed to read partition with ID: {} for stream with ID: {} for topic with ID: {} and path: {}",
- topic.topic_id, topic.stream_id, topic.topic_id, &topic.partitions_path))
- .map_err(|_| IggyError::CannotReadPartitions)?;
-
- let mut unloaded_partitions = Vec::new();
- while let Some(dir_entry) = dir_entries.next().transpose().unwrap_or(None) {
- let metadata = dir_entry.metadata();
- if metadata.is_err() || metadata.unwrap().is_file() {
- continue;
- }
-
- let name = dir_entry.file_name().into_string().unwrap();
- let partition_id = name.parse::<u32>();
- if partition_id.is_err() {
- error!("Invalid partition ID file with name: '{}'.", name);
- continue;
- }
-
- let partition_id = partition_id.unwrap();
- let partition_state = state.partitions.get(&partition_id);
- if partition_state.is_none() {
- let stream_id = topic.stream_id;
- let topic_id = topic.topic_id;
- error!(
- "Partition with ID: '{partition_id}' for stream with ID: '{stream_id}' and topic with ID: '{topic_id}' was not found in state, but exists on disk and will be removed."
- );
- if let Err(error) = fs_utils::remove_dir_all(&dir_entry.path()).await {
- error!("Cannot remove partition directory: {error}");
- } else {
- warn!(
- "Partition with ID: '{partition_id}' for stream with ID: '{stream_id}' and topic with ID: '{topic_id}' was removed."
- );
- }
- continue;
- }
-
- let partition_state = partition_state.unwrap();
- let partition = Partition::create(
- topic.stream_id,
- topic.topic_id,
- partition_id,
- false,
- topic.config.clone(),
- topic.storage.clone(),
- message_expiry,
- topic.messages_count_of_parent_stream.clone(),
- topic.messages_count.clone(),
- topic.size_of_parent_stream.clone(),
- topic.size_bytes.clone(),
- topic.segments_count_of_parent_stream.clone(),
- partition_state.created_at,
- );
- unloaded_partitions.push(partition);
- }
-
- let state_partition_ids = state.partitions.keys().copied().collect::<AHashSet<u32>>();
- let unloaded_partition_ids = unloaded_partitions
- .iter()
- .map(|partition| partition.partition_id)
- .collect::<AHashSet<u32>>();
- let missing_ids = state_partition_ids
- .difference(&unloaded_partition_ids)
- .copied()
- .collect::<AHashSet<u32>>();
- if missing_ids.is_empty() {
- info!(
- "All partitions for topic with ID: '{}' for stream with ID: '{}' found on disk were found in state.",
- topic.topic_id, topic.stream_id
- );
- } else {
- warn!(
- "Partitions with IDs: '{missing_ids:?}' for topic with ID: '{topic_id}' for stream with ID: '{stream_id}' were not found on disk.",
- topic_id = topic.topic_id,
- stream_id = topic.stream_id
- );
- if topic.config.recovery.recreate_missing_state {
- info!(
- "Recreating missing state in recovery config is enabled, missing partitions will be created for topic with ID: '{}' for stream with ID: '{}'.",
- topic.topic_id, topic.stream_id
- );
-
- for partition_id in missing_ids {
- let partition_state = state.partitions.get(&partition_id).unwrap();
- let mut partition = Partition::create(
- topic.stream_id,
- topic.topic_id,
- partition_id,
- true,
- topic.config.clone(),
- topic.storage.clone(),
- message_expiry,
- topic.messages_count_of_parent_stream.clone(),
- topic.messages_count.clone(),
- topic.size_of_parent_stream.clone(),
- topic.size_bytes.clone(),
- topic.segments_count_of_parent_stream.clone(),
- partition_state.created_at,
- );
- partition.persist().await.with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to persist partition: {partition}"
- )
- })?;
- partition.segments.clear();
- unloaded_partitions.push(partition);
- info!(
- "Created missing partition with ID: '{partition_id}', for topic with ID: '{}' for stream with ID: '{}'.",
- topic.topic_id, topic.stream_id
- );
- }
- } else {
- warn!(
- "Recreating missing state in recovery config is disabled, missing partitions will not be created for topic with ID: '{}' for stream with ID: '{}'.",
- topic.topic_id, topic.stream_id
- );
- }
- }
-
- let stream_id = topic.stream_id;
- let topic_id = topic.topic_id;
- let loaded_partitions = Arc::new(Mutex::new(Vec::new()));
- let mut load_partitions = Vec::new();
- for mut partition in unloaded_partitions {
- let loaded_partitions = loaded_partitions.clone();
- let partition_state = state.partitions.remove(&partition.partition_id).unwrap();
- let load_partition = compio::runtime::spawn(async move {
- match partition.load(partition_state).await {
- Ok(_) => {
- loaded_partitions.lock().await.push(partition);
- }
- Err(error) => {
- error!(
- "Failed to load partition with ID: {} for stream with ID: {stream_id} and topic with ID: {topic_id}. Error: {error}",
- partition.partition_id
- );
- }
- }
- });
- load_partitions.push(load_partition);
- }
-
- join_all(load_partitions).await;
- for partition in loaded_partitions.lock().await.drain(..) {
- topic
- .partitions
- .insert(partition.partition_id, IggyRwLock::new(partition));
- }
-
- for consumer_group in state.consumer_groups.into_values() {
- let consumer_group = ConsumerGroup::new(
- topic.topic_id,
- consumer_group.id,
- &consumer_group.name,
- topic.get_partitions_count(),
- );
- topic
- .consumer_groups_ids
- .insert(consumer_group.name.to_owned(), consumer_group.group_id);
- topic
- .consumer_groups
- .borrow_mut()
- .insert(consumer_group.group_id, consumer_group);
- }
-
- info!("Loaded topic {topic}");
-
- Ok(())
+ todo!();
}
async fn save(&self, topic: &Topic) -> Result<(), IggyError> {
diff --git a/core/server/src/streaming/topics/topic.rs b/core/server/src/streaming/topics/topic.rs
index b4fcba69..7fcb4831 100644
--- a/core/server/src/streaming/topics/topic.rs
+++ b/core/server/src/streaming/topics/topic.rs
@@ -133,8 +133,8 @@ impl Topic {
consumer_groups_ids: AHashMap::new(),
current_consumer_group_id: Arc::new(AtomicU32::new(1)),
current_partition_id: Arc::new(AtomicU32::new(1)),
- message_expiry: Topic::get_message_expiry(message_expiry, &config),
- max_topic_size: Topic::get_max_topic_size(max_topic_size, &config)?,
+ message_expiry: message_expiry,
+ max_topic_size: max_topic_size,
compression_algorithm,
replication_factor,
config,
@@ -228,32 +228,6 @@ impl Topic {
}
}
}
-
- pub fn get_max_topic_size(
- max_topic_size: MaxTopicSize,
- config: &SystemConfig,
- ) -> Result<MaxTopicSize, IggyError> {
- match max_topic_size {
- MaxTopicSize::ServerDefault => Ok(config.topic.max_size),
- _ => {
- if max_topic_size.as_bytes_u64() < config.segment.size.as_bytes_u64() {
- Err(IggyError::InvalidTopicSize(
- max_topic_size,
- config.segment.size,
- ))
- } else {
- Ok(max_topic_size)
- }
- }
- }
- }
-
- pub fn get_message_expiry(message_expiry: IggyExpiry, config: &SystemConfig) -> IggyExpiry {
- match message_expiry {
- IggyExpiry::ServerDefault => config.segment.message_expiry,
- _ => message_expiry,
- }
- }
}
impl Sizeable for Topic {
diff --git a/core/server/src/streaming/topics/topic2.rs b/core/server/src/streaming/topics/topic2.rs
index 01586a87..3f930990 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -1,8 +1,9 @@
+use crate::configs::system::SystemConfig;
use crate::slab::topics;
use crate::slab::traits_ext::{EntityMarker, IntoComponents, IntoComponentsById};
use crate::slab::{Keyed, consumer_groups::ConsumerGroups, partitions::Partitions};
use crate::streaming::stats::stats::{PartitionStats, TopicStats};
-use iggy_common::{CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize};
+use iggy_common::{CompressionAlgorithm, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize};
use slab::Slab;
use std::cell::{Ref, RefMut};
use std::sync::Arc;
@@ -160,7 +161,7 @@ impl TopicRoot {
Self {
id: 0,
name,
- created_at: IggyTimestamp::now(),
+ created_at,
replication_factor,
message_expiry,
compression_algorithm: compression,
@@ -193,6 +194,10 @@ impl TopicRoot {
self.message_expiry
}
+ pub fn max_topic_size(&self) -> MaxTopicSize {
+ self.max_topic_size
+ }
+
pub fn id(&self) -> usize {
self.id
}
@@ -243,4 +248,42 @@ impl TopicRoot {
topic.id = idx;
idx
}
+
+ pub fn created_at(&self) -> IggyTimestamp {
+ self.created_at
+ }
+
+ pub fn compression_algorithm(&self) -> CompressionAlgorithm {
+ self.compression_algorithm
+ }
+
+ pub fn replication_factor(&self) -> u8 {
+ self.replication_factor
+ }
+}
+
+pub fn resolve_max_topic_size(
+ max_topic_size: MaxTopicSize,
+ config: &SystemConfig,
+) -> Result<MaxTopicSize, IggyError> {
+ match max_topic_size {
+ MaxTopicSize::ServerDefault => Ok(config.topic.max_size),
+ _ => {
+ if max_topic_size.as_bytes_u64() < config.segment.size.as_bytes_u64() {
+ Err(IggyError::InvalidTopicSize(
+ max_topic_size,
+ config.segment.size,
+ ))
+ } else {
+ Ok(max_topic_size)
+ }
+ }
+ }
+}
+
+pub fn resolve_message_expiry(message_expiry: IggyExpiry, config: &SystemConfig) -> IggyExpiry {
+ match message_expiry {
+ IggyExpiry::ServerDefault => config.segment.message_expiry,
+ _ => message_expiry,
+ }
}