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/12 18:10:10 UTC
(iggy) 02/02: finito
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch implement_new_arch_for_topic
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 3f8fbd3764e3bcfdd7372be652e8ae0768491be6
Author: numminex <g....@gmail.com>
AuthorDate: Tue Aug 12 20:09:56 2025 +0200
finito
---
.../handlers/streams/create_stream_handler.rs | 19 +-
.../handlers/streams/delete_stream_handler.rs | 5 +-
.../binary/handlers/topics/create_topic_handler.rs | 30 +--
.../binary/handlers/topics/delete_topic_handler.rs | 1 +
core/server/src/shard/mod.rs | 53 ++---
core/server/src/shard/system/consumer_groups.rs | 19 +-
core/server/src/shard/system/partitions.rs | 30 +--
core/server/src/shard/system/streams.rs | 117 ++++-------
core/server/src/shard/system/topics.rs | 178 +++++++---------
core/server/src/shard/system/utils.rs | 9 +-
core/server/src/shard/transmission/event.rs | 14 +-
core/server/src/slab/mod.rs | 2 +-
core/server/src/slab/partitions.rs | 121 ++++-------
core/server/src/slab/streams.rs | 233 ++++++++++++++-------
core/server/src/slab/topics.rs | 191 +++++++++++++++--
core/server/src/slab/traits_ext.rs | 176 +++++++++-------
core/server/src/streaming/partitions/partition2.rs | 31 +--
core/server/src/streaming/segments/segment2.rs | 2 +-
core/server/src/streaming/streams/stream2.rs | 170 +++++++++++----
core/server/src/streaming/topics/topic2.rs | 99 ++++++---
20 files changed, 892 insertions(+), 608 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 4461745e..0d374297 100644
--- a/core/server/src/binary/handlers/streams/create_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/create_stream_handler.rs
@@ -23,6 +23,7 @@ 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::state::command::EntryCommand;
use crate::state::models::CreateStreamWithId;
use crate::streaming::session::Session;
@@ -51,30 +52,22 @@ impl ServerCommandHandler for CreateStream {
debug!("session: {session}, command: {self}");
let stream_id = self.stream_id;
let name = self.name.clone();
- let stats = Arc::new(StreamStats::new());
- let new_stream_id = shard
- .create_stream2(session, stream_id, self.name.clone(), stats.clone())
+ let stream = shard
+ .create_stream2(session, stream_id, self.name.clone())
.await?;
shard_info!(
shard.id,
"Created stream with new API, Stream ID: {}, name: '{}'.",
- new_stream_id,
+ stream.id(),
name
);
let event = ShardEvent::CreatedStream2 {
- id: new_stream_id,
- name: self.name.clone(),
- stats,
+ id: stream.id(),
+ stream
};
let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
- //TODO: Replace the mapping from line 89 with this once the Stream layer is finished.
- let _ = shard.streams2.with_stream_by_id(
- &Identifier::numeric(new_stream_id as u32).unwrap(),
- |stream| mapper::map_stream2(stream),
- );
-
let created_stream_id = shard
.create_stream(session, stream_id, &name)
.await
diff --git a/core/server/src/binary/handlers/streams/delete_stream_handler.rs b/core/server/src/binary/handlers/streams/delete_stream_handler.rs
index af12beda..c7b9ad2c 100644
--- a/core/server/src/binary/handlers/streams/delete_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/delete_stream_handler.rs
@@ -23,6 +23,7 @@ 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::partitions::partition;
use crate::streaming::session::Session;
@@ -58,8 +59,8 @@ impl ServerCommandHandler for DeleteStream {
})?;
shard_info!(
shard.id,
- "Deleted stream2 with name: {}, ID: {}",
- stream2.name(),
+ "Deleted stream with name: {}, ID: {}",
+ stream2.root().name(),
stream2.id()
);
let event = ShardEvent::DeletedStream2 {
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 848a4c90..3f913685 100644
--- a/core/server/src/binary/handlers/topics/create_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/create_topic_handler.rs
@@ -24,6 +24,7 @@ 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::EntityMarker;
use crate::state::command::EntryCommand;
use crate::state::models::CreateTopicWithId;
use crate::state::system::TopicState;
@@ -55,12 +56,7 @@ impl ServerCommandHandler for CreateTopic {
debug!("session: {session}, command: {self}");
let stream_id = self.stream_id.clone();
let maybe_topic_id = self.topic_id;
- //shard.ensure_stream_exists(&stream_id)?;
- let parent = shard
- .streams2
- .with_stats_by_id(&stream_id, |stats| stats.clone());
- let stats = Arc::new(TopicStats::new(parent));
- let new_topic_id = shard
+ let topic = shard
.create_topic2(
session,
&stream_id,
@@ -70,30 +66,24 @@ impl ServerCommandHandler for CreateTopic {
self.compression_algorithm,
self.max_topic_size,
self.replication_factor,
- stats.clone(),
)
.await?;
+ let topic_id = topic.id();
+ // Send events for topic creation.
+ let event = ShardEvent::CreatedTopic2 {
+ stream_id: self.stream_id.clone(),
+ topic,
+ };
+ let _responses = shard.broadcast_event_to_all_shards(event).await;
shard
.create_partitions2(
session,
&stream_id,
- &Identifier::numeric(new_topic_id as u32).unwrap(),
+ &Identifier::numeric(topic_id as u32).unwrap(),
self.partitions_count,
)
.await?;
-
- let event = ShardEvent::CreatedTopic2 {
- stream_id: self.stream_id.clone(),
- id: new_topic_id,
- 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,
- stats,
- };
- let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
let (topic_id, partition_ids) = shard
.create_topic(
session,
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 af16cf8c..70fccd8c 100644
--- a/core/server/src/binary/handlers/topics/delete_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
@@ -24,6 +24,7 @@ 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;
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index c7994355..5ef6d36c 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -68,7 +68,7 @@ use crate::{
},
},
shard_error, shard_info, shard_warn,
- slab::streams::Streams,
+ slab::{streams::Streams, traits_ext::EntityMarker},
state::{
StateKind,
system::{StreamState, SystemState, UserState},
@@ -631,7 +631,7 @@ impl IggyShard {
partition_ids,
} => {
self.delete_partitions2_bypass_auth(
- &stream_id,
+ &stream_id,
&topic_id,
partitions_count,
partition_ids,
@@ -667,11 +667,13 @@ impl IggyShard {
"{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}"
)
})?;
- let topic = stream.get_topic_mut(&topic_id).with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id}"
- )
- })?;
+ let topic = stream
+ .get_topic_mut(&topic_id)
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id}"
+ )
+ })?;
topic.reassign_consumer_groups();
if partitions.len() > 0 {
self.metrics.decrement_partitions(partitions_count as u32);
@@ -848,10 +850,8 @@ impl IggyShard {
self.tcp_bound_address.set(Some(address));
Ok(())
}
- ShardEvent::CreatedStream2 { id, name, stats } => {
- let stream_id = self
- .create_stream2_bypass_auth(name.to_owned(), stats.clone())
- .await?;
+ ShardEvent::CreatedStream2 { id, stream } => {
+ let stream_id = self.create_stream2_bypass_auth(stream);
assert_eq!(stream_id, id);
Ok(())
}
@@ -861,25 +861,10 @@ impl IggyShard {
Ok(())
}
ShardEvent::CreatedTopic2 {
- id,
stream_id,
- name,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor,
- stats,
+ topic,
} => {
- let topic_id = self.create_topic2_bypass_auth(
- &stream_id,
- name.to_owned(),
- replication_factor,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- stats.clone(),
- )?;
- assert_eq!(topic_id, id);
+ let _topic_id = self.create_topic2_bypass_auth(&stream_id, topic)?;
Ok(())
}
ShardEvent::CreatedPartitions2 {
@@ -895,7 +880,9 @@ impl IggyShard {
stream_id,
topic_id,
} => {
- let topic = self.delete_topic_bypass_auth2(&stream_id, &topic_id).await?;
+ let topic = self
+ .delete_topic_bypass_auth2(&stream_id, &topic_id)
+ .await?;
assert_eq!(topic.id(), id);
Ok(())
}
@@ -916,7 +903,7 @@ impl IggyShard {
compression_algorithm,
max_topic_size,
replication_factor,
- );
+ )?;
Ok(())
}
ShardEvent::CreatedConsumerGroup2 {
@@ -929,7 +916,8 @@ impl IggyShard {
let id = self.create_consumer_group_bypass_auth2(
&stream_id,
&topic_id,
- members.clone(), name.clone(),
+ members.clone(),
+ name.clone(),
)?;
assert_eq!(id, cg_id);
Ok(())
@@ -940,7 +928,8 @@ impl IggyShard {
topic_id,
group_id,
} => {
- let cg = self.delete_consumer_group_bypass_auth2(&stream_id, &topic_id, &group_id)?;
+ let cg =
+ self.delete_consumer_group_bypass_auth2(&stream_id, &topic_id, &group_id)?;
assert_eq!(cg.id(), id);
Ok(())
}
diff --git a/core/server/src/shard/system/consumer_groups.rs b/core/server/src/shard/system/consumer_groups.rs
index e58f36df..2283fbb4 100644
--- a/core/server/src/shard/system/consumer_groups.rs
+++ b/core/server/src/shard/system/consumer_groups.rs
@@ -119,10 +119,10 @@ impl IggyShard {
{
let topic_id = self
.streams2
- .with_topic_by_id(stream_id, topic_id, |topic| topic.id());
+ .with_topic_root_by_id(stream_id, topic_id, |topic| topic.id());
let stream_id = self
.streams2
- .with_stream_by_id(stream_id, |stream| stream.id());
+ .with_root_by_id(stream_id, |stream| stream.id());
self.permissioner.borrow().create_consumer_group(
session.get_user_id(),
stream_id as u32,
@@ -151,7 +151,7 @@ impl IggyShard {
) -> Result<usize, IggyError> {
let id = self
.streams2
- .with_topic_by_id_mut(stream_id, topic_id, |topic| {
+ .with_topic_root_by_id_mut(stream_id, topic_id, |topic| {
let partitions = topic.partitions().with(|partitions| {
let (info, _, _, _, _, _) = partitions.into_components();
info.iter()
@@ -233,10 +233,10 @@ impl IggyShard {
{
let topic_id = self
.streams2
- .with_topic_by_id(stream_id, topic_id, |topic| topic.id());
+ .with_topic_root_by_id(stream_id, topic_id, |topic| topic.id());
let stream_id = self
.streams2
- .with_stream_by_id(stream_id, |stream| stream.id());
+ .with_root_by_id(stream_id, |stream| stream.id());
self.permissioner.borrow().delete_consumer_group(
session.get_user_id(),
stream_id as u32,
@@ -263,20 +263,19 @@ impl IggyShard {
) -> Result<consumer_group2::ConsumerGroup, IggyError> {
let cg = self
.streams2
- .with_topic_by_id_mut(stream_id, topic_id, |topic| {
+ .with_topic_root_by_id_mut(stream_id, topic_id, |root| {
match group_id.kind {
iggy_common::IdKind::Numeric => {
- topic.consumer_groups_mut().with_mut(|container| {
+ root.consumer_groups_mut().with_mut(|container| {
container.try_remove(group_id.get_u32_value().unwrap() as usize)
})
}
iggy_common::IdKind::String => {
let key = group_id.get_string_value().unwrap();
- let id = topic
+ let id = root
.consumer_groups()
.with_index(|index| *(index.get(&key).unwrap()));
- topic
- .consumer_groups_mut()
+ root.consumer_groups_mut()
.with_mut(|container| container.try_remove(id))
}
}
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index 2059d25c..3cc1ed61 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -24,6 +24,7 @@ use crate::configs::system::SystemConfig;
use crate::shard::IggyShard;
use crate::slab::traits_ext::Delete;
use crate::slab::traits_ext::EntityComponentSystem;
+use crate::slab::traits_ext::EntityMarker;
use crate::slab::traits_ext::Insert;
use crate::streaming::deduplication::message_deduplicator::MessageDeduplicator;
use crate::streaming::partitions::partition::Partition;
@@ -74,12 +75,16 @@ impl IggyShard {
partitions_count: u32,
) -> Result<Vec<partition2::Partition>, IggyError> {
self.ensure_authenticated(session)?;
+ /*
+ self.ensure_stream_exists(stream_id)?;
+ self.ensure_topic_exists(stream_id, topic_id)?;
+ */
let numeric_stream_id =
self.streams2
- .with_stream_by_id(stream_id, |stream| stream.id()) as u32;
+ .with_root_by_id(stream_id, |stream| stream.id()) as u32;
let numeric_topic_id =
self.streams2
- .with_topic_by_id(stream_id, topic_id, |topic| topic.id()) as u32;
+ .with_topic_root_by_id(stream_id, topic_id, |topic| topic.id()) as u32;
self.validate_partition_permissions(
session,
@@ -88,10 +93,9 @@ impl IggyShard {
"create",
)?;
- let parent_stats = self.streams2.with_stream_by_id(stream_id, |stream| {
- stream
- .topics()
- .with_topic_stats_by_id(topic_id, |stats| stats)
+ let parent_stats = self.streams2.with_root_by_id(stream_id, |root| {
+ root.topics()
+ .with_stats_by_id(topic_id, |stats| stats.clone())
});
let partitions = self.create_and_insert_partitions_mem(
stream_id,
@@ -172,8 +176,8 @@ impl IggyShard {
partition: partition2::Partition,
) -> usize {
self.streams2
- .with_topic_by_id_mut(stream_id, topic_id, |topic| {
- topic.partitions_mut().insert(partition)
+ .with_partitions_mut(stream_id, topic_id, |partitions| {
+ partitions.insert(partition)
})
}
@@ -265,10 +269,10 @@ impl IggyShard {
let numeric_stream_id =
self.streams2
- .with_stream_by_id(stream_id, |stream| stream.id()) as u32;
+ .with_root_by_id(stream_id, |stream| stream.id()) as u32;
let numeric_topic_id =
self.streams2
- .with_topic_by_id(stream_id, topic_id, |topic| topic.id()) as u32;
+ .with_topic_root_by_id(stream_id, topic_id, |topic| topic.id()) as u32;
self.validate_partition_permissions(
session,
@@ -290,13 +294,13 @@ impl IggyShard {
) -> Vec<u32> {
let deleted_partition_ids =
self.streams2
- .with_topic_by_id_mut(stream_id, topic_id, |topic| {
+ .with_topic_root_by_id_mut(stream_id, topic_id, |topic| {
let partitions = topic.partitions_mut();
- let current_count = partitions.count() as u32;
+ let current_count = partitions.len() as u32;
let partitions_to_delete = partitions_count.min(current_count);
let start_idx = (current_count - partitions_to_delete) as usize;
let mut deleted_ids = Vec::with_capacity(partitions_to_delete as usize);
- for idx in (start_idx..current_count as usize) {
+ for idx in start_idx..current_count as usize {
let partition = topic.partitions_mut().delete(idx);
assert_eq!(partition.id(), idx);
deleted_ids.push(partition.id() as u32);
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index 958cab48..b76d1025 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -19,6 +19,9 @@
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::streaming::session::Session;
use crate::streaming::stats::stats::StreamStats;
@@ -28,7 +31,7 @@ use crate::streaming::streams::stream2;
use error_set::ErrContext;
use futures::future::try_join_all;
use iggy_common::locking::IggyRwLockFn;
-use iggy_common::{IdKind, Identifier, IggyError};
+use iggy_common::{IdKind, Identifier, IggyError, IggyTimestamp};
use std::cell::{Ref, RefCell, RefMut};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
@@ -213,45 +216,39 @@ impl IggyShard {
session: &Session,
stream_id: Option<u32>,
name: String,
- stats: Arc<StreamStats>,
- ) -> Result<usize, IggyError> {
+ ) -> Result<stream2::Stream, IggyError> {
self.ensure_authenticated(session)?;
self.permissioner
.borrow()
.create_stream(session.get_user_id())?;
- let stream_id = self.create_stream2_base(name, stats).await?;
- create_stream_file_hierarchy(self.id, stream_id, &self.config.system).await?;
- Ok(stream_id)
+ let exists = self.streams2.exists(
+ &Identifier::from_str_value(&name).unwrap());
+
+ if exists {
+ return Err(IggyError::StreamNameAlreadyExists(
+ name,
+ ));
+ }
+ let stream = self.create_and_insert_stream_mem(name);
+ create_stream_file_hierarchy(self.id, stream.id(), &self.config.system).await?;
+ Ok(stream)
}
- pub async fn create_stream2_bypass_auth(
- &self,
- name: String,
- stats: Arc<StreamStats>,
- ) -> Result<usize, IggyError> {
- self.create_stream2_base(name, stats).await
+ fn create_and_insert_stream_mem(&self, name: String) -> stream2::Stream {
+ let now = IggyTimestamp::now();
+ let stats = Arc::new(StreamStats::new());
+ let mut stream = stream2::Stream::new(name, stats, now);
+ let id = self.insert_stream_mem(stream.clone());
+ stream.update_id(id);
+ stream
}
- async fn create_stream2_base(
- &self,
- name: String,
- stats: Arc<StreamStats>,
- ) -> Result<usize, IggyError> {
- let exists = self.streams2.exists(&Identifier::named(&name).unwrap());
- if exists {
- return Err(IggyError::StreamNameAlreadyExists(name));
- }
- let key = name.clone();
- let stream = stream2::Stream::new(name);
- let stream_id = self
- .streams2
- .with_mut(|streams| stream.insert_into(streams));
- self.streams2.with_index_mut(|index| {
- index.insert(key, stream_id);
- });
- self.streams2
- .with_stats_mut(|container| container.insert(stats));
- Ok(stream_id)
+ fn insert_stream_mem(&self, stream: stream2::Stream) -> usize {
+ self.streams2.insert(stream)
+ }
+
+ pub fn create_stream2_bypass_auth(&self, stream: stream2::Stream) -> usize {
+ self.insert_stream_mem(stream)
}
pub async fn create_stream(
@@ -405,9 +402,7 @@ impl IggyShard {
name: &str,
) -> Result<(), IggyError> {
self.ensure_authenticated(session)?;
- let stream_id = self
- .streams2
- .with_stream_by_id(id, |stream| stream.id() as u32);
+ let stream_id = self.streams2.with_root_by_id(id, |root| root.id() as u32);
self.permissioner
.borrow()
@@ -426,7 +421,7 @@ impl IggyShard {
fn update_stream2_base(&self, id: &Identifier, name: String) -> Result<String, IggyError> {
let old_name = self
.streams2
- .with_stream_by_id(id, |stream| stream.name().clone());
+ .with_root_by_id(id, |root| root.name().clone());
if old_name == name {
return Ok(old_name);
@@ -437,10 +432,10 @@ impl IggyShard {
return Err(IggyError::StreamNameAlreadyExists(name.to_string()));
}
- self.streams2.with_stream_by_id_mut(id, |stream| {
- stream.set_name(name.clone());
+ self.streams2.with_root_by_id_mut(id, |root| {
+ root.set_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");
@@ -519,39 +514,21 @@ impl IggyShard {
}
fn delete_stream2_base(&self, id: &Identifier) -> Result<stream2::Stream, IggyError> {
- let (stream_id, stream_name) = self
- .streams2
- .with_stream_by_id(id, |stream| (stream.id() as u32, stream.name().clone()));
-
- let stream = self.streams2.with_mut(|streams| {
- streams
- .try_remove(stream_id as usize)
- .ok_or_else(|| match id.kind {
- iggy_common::IdKind::Numeric => IggyError::StreamIdNotFound(stream_id as u32),
- iggy_common::IdKind::String => {
- IggyError::StreamNameNotFound(stream_name.clone())
- }
- })
- })?;
+ let id = self.streams2.get_index(id);
+ let stream = self.streams2.delete(id);
+ let stats = stream.stats();
- self.streams2.with_stats_mut(|stats| {
- if let Some(stream_stats) = stats.try_remove(stream_id as usize) {
- 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(stream_stats.messages_count());
- self.metrics
- .decrement_segments(stream_stats.segments_count());
- } else {
- self.metrics.decrement_streams(1);
- }
- });
+ 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.client_manager
.borrow_mut()
.delete_consumer_groups_for_stream(stream_id as u32);
-
+ */
Ok(stream)
}
@@ -562,9 +539,7 @@ impl IggyShard {
) -> Result<stream2::Stream, IggyError> {
self.ensure_authenticated(session)?;
// self.ensure_stream_exists(id)?;
- let stream_id = self
- .streams2
- .with_stream_by_id(id, |stream| stream.id() as u32);
+ let stream_id = self.streams2.with_root_by_id(id, |root| root.id() as u32);
self.permissioner
.borrow()
.delete_stream(session.get_user_id(), stream_id as u32)
@@ -631,7 +606,7 @@ impl IggyShard {
// self.ensure_stream_exists(id)?;
let stream_id = self
.streams2
- .with_stream_by_id(id, |stream| stream.id() as u32);
+ .with_root_by_id(id, |root| root.id() as u32);
self.permissioner
.borrow()
.purge_stream(session.get_user_id(), stream_id)
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index ff327975..d0beaa31 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -24,10 +24,14 @@ 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::streaming::session::Session;
-use crate::streaming::stats::stats::TopicStats;
+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;
@@ -35,7 +39,9 @@ use crate::streaming::topics::topic2;
use clap::Id;
use error_set::ErrContext;
use iggy_common::locking::IggyRwLockFn;
-use iggy_common::{CompressionAlgorithm, Identifier, IggyError, IggyExpiry, MaxTopicSize};
+use iggy_common::{
+ CompressionAlgorithm, Identifier, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize,
+};
use tokio_util::io::StreamReader;
use tracing::info;
@@ -152,13 +158,10 @@ impl IggyShard {
compression: CompressionAlgorithm,
max_topic_size: MaxTopicSize,
replication_factor: Option<u8>,
- stats: Arc<TopicStats>,
- ) -> Result<usize, IggyError> {
+ ) -> Result<topic2::Topic, IggyError> {
self.ensure_authenticated(session)?;
//self.ensure_stream_exists(stream_id)?;
- let numeric_stream_id = self
- .streams2
- .with_stream_by_id(stream_id, |stream| stream.id());
+ let numeric_stream_id = self.streams2.with_root_by_id(stream_id, |root| root.id());
{
self.permissioner
.borrow()
@@ -170,94 +173,78 @@ impl IggyShard {
)
})?;
}
- let topic_id = self.create_topic2_base(
+ let exists = self.streams2.with_topics(stream_id, |topics| {
+ topics.exists(&Identifier::from_str_value(&name).unwrap())
+ });
+ if exists {
+ return Err(IggyError::TopicNameAlreadyExists(
+ name,
+ numeric_stream_id as u32,
+ ));
+ }
+
+ let parent_stats = self
+ .streams2
+ .with_stats_by_id(stream_id, |stats| stats.clone());
+ let topic = self.create_and_insert_topics_mem(
stream_id,
name,
replication_factor.unwrap_or(1),
message_expiry,
compression,
max_topic_size,
- stats,
- )?;
-
- self.streams2.with_topic_by_id(
- stream_id,
- &Identifier::numeric(topic_id as u32).unwrap(),
- |topic| {
- let message_expiry = match topic.message_expiry() {
- IggyExpiry::ServerDefault => self.config.system.segment.message_expiry,
- _ => message_expiry,
- };
- shard_info!(self.id, "Topic message expiry: {}", message_expiry);
- },
+ 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)
+ create_topic_file_hierarchy(self.id, numeric_stream_id, topic.id(), &self.config.system)
.await?;
- Ok(topic_id)
+ Ok(topic)
}
- pub fn create_topic2_bypass_auth(
+ fn create_and_insert_topics_mem(
&self,
stream_id: &Identifier,
name: String,
- replication_factor: Option<u8>,
+ replication_factor: u8,
message_expiry: IggyExpiry,
compression: CompressionAlgorithm,
max_topic_size: MaxTopicSize,
- stats: Arc<TopicStats>,
- ) -> Result<usize, IggyError> {
- let topic_id = self.create_topic2_base(
- stream_id,
+ parent_stats: Arc<StreamStats>,
+ ) -> topic2::Topic {
+ let stats = Arc::new(TopicStats::new(parent_stats));
+ let now = IggyTimestamp::now();
+ let mut topic = topic2::Topic::new(
name,
- replication_factor.unwrap_or(1),
+ stats,
+ now,
+ replication_factor,
message_expiry,
compression,
max_topic_size,
- stats,
- )?;
- Ok(topic_id)
+ );
+
+ let id = self.insert_topic_mem(stream_id, topic.clone());
+ topic.update_id(id);
+ topic
+ }
+
+ 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))
}
- fn create_topic2_base(
+ pub fn create_topic2_bypass_auth(
&self,
stream_id: &Identifier,
- name: String,
- replication_factor: u8,
- message_expiry: IggyExpiry,
- compression: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- stats: Arc<TopicStats>,
+ topic: topic2::Topic,
) -> Result<usize, IggyError> {
- let topic_id = self.streams2.with_stream_by_id(stream_id, |stream| {
- let exists = stream.topics().exists(&Identifier::named(&name).unwrap());
- if exists {
- // TODO: Fixme, replace the second argument with identifier, rather than numeric ID.
- return Err(IggyError::TopicNameAlreadyExists(name.to_owned(), 0));
- }
- let topic = topic2::Topic::new(
- name,
- replication_factor,
- message_expiry,
- compression,
- max_topic_size,
- );
- let name = topic.name().clone();
- let topic_id = stream.topics().with_mut(|topics| {
- let topic_id = topic.insert_into(topics);
- topic_id
- });
- stream.topics().with_mut_index(|index| {
- index.insert(name, topic_id);
- });
-
- stream.topics().with_stats_mut(|container| {
- container.insert(stats);
- });
- Ok(topic_id)
- })?;
-
+ let topic_id = self.insert_topic_mem(stream_id, topic);
Ok(topic_id)
}
@@ -390,10 +377,10 @@ impl IggyShard {
{
let topic_id = self
.streams2
- .with_topic_by_id(stream_id, topic_id, |topic| topic.id());
+ .with_topic_root_by_id(stream_id, topic_id, |topic| topic.id());
let stream_id = self
.streams2
- .with_stream_by_id(stream_id, |stream| stream.id());
+ .with_root_by_id(stream_id, |stream| stream.id());
self.permissioner.borrow().update_topic(
session.get_user_id(),
stream_id as u32,
@@ -407,6 +394,17 @@ impl IggyShard {
)
})?;
}
+ 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,
topic_id,
@@ -453,7 +451,7 @@ impl IggyShard {
) {
let (old_name, new_name) =
self.streams2
- .with_topic_by_id_mut(stream_id, topic_id, |topic| {
+ .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);
@@ -465,7 +463,7 @@ impl IggyShard {
});
if old_name != new_name {
self.streams2.with_topics(stream_id, |topics| {
- topics.with_mut_index(|index| {
+ 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);
@@ -600,10 +598,10 @@ impl IggyShard {
{
let topic_id = self
.streams2
- .with_topic_by_id(stream_id, topic_id, |topic| topic.id());
+ .with_topic_root_by_id(stream_id, topic_id, |topic| topic.id());
let stream_id = self
.streams2
- .with_stream_by_id(stream_id, |stream| stream.id());
+ .with_root_by_id(stream_id, |stream| stream.id());
self.permissioner
.borrow()
.delete_topic(session.get_user_id(), stream_id as u32, topic_id as u32)
@@ -617,7 +615,7 @@ impl IggyShard {
let topic = self.delete_topic_base2(stream_id, topic_id).with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to delete topic with ID: {topic_id} in stream with ID: {stream_id}")
})?;
- // TODO: Remove partitions.
+ // TODO: Decrease the stats
Ok(topic)
}
@@ -635,26 +633,12 @@ impl IggyShard {
stream_id: &Identifier,
topic_id: &Identifier,
) -> Result<topic2::Topic, IggyError> {
- let topic = self.streams2.with_stream_by_id(stream_id, |stream| {
- let id = stream
- .topics()
- .with_topic_by_id(topic_id, |topic| topic.id());
- stream.topics().with_mut(|container| {
- container.try_remove(id).ok_or_else(|| {
- let topic_name = stream
- .topics()
- .with_topic_by_id(topic_id, |topic| topic.name().clone());
- IggyError::TopicNameNotFound(topic_name, stream.name().clone())
- })
- })
- })?;
- let id = topic.id();
- self.streams2.with_topics(stream_id, |topics| {
- topics.with_stats_mut(|container| {
- container
- .try_remove(id)
- .expect("Topic delete: topic stats not found");
- });
+ 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
});
Ok(topic)
}
@@ -727,10 +711,10 @@ impl IggyShard {
{
let topic_id = self
.streams2
- .with_topic_by_id(stream_id, topic_id, |topic| topic.id());
+ .with_topic_root_by_id(stream_id, topic_id, |topic| topic.id());
let stream_id = self
.streams2
- .with_stream_by_id(stream_id, |stream| stream.id());
+ .with_root_by_id(stream_id, |stream| stream.id());
self.permissioner.borrow().purge_topic(
session.get_user_id(),
stream_id as u32,
diff --git a/core/server/src/shard/system/utils.rs b/core/server/src/shard/system/utils.rs
index 6ad3b224..28db67dc 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;
+use crate::{shard::IggyShard, slab::traits_ext::EntityComponentSystem};
impl IggyShard {
pub fn ensure_stream_exists(&self, stream_id: &Identifier) -> Result<(), IggyError> {
@@ -16,9 +16,10 @@ impl IggyShard {
topic_id: &Identifier,
) -> Result<(), IggyError> {
//self.ensure_stream_exists(stream_id)?;
+ let stream_id = self.streams2.get_index(stream_id);
let exists = self
.streams2
- .with_stream_by_id(stream_id, |stream| stream.topics().exists(topic_id));
+ .with_by_id(stream_id, |(root, _)| root.topics().exists(topic_id));
if !exists {
return Err(IggyError::TopicIdNotFound(0, 0));
}
@@ -35,8 +36,8 @@ impl IggyShard {
//self.ensure_topic_exists(stream_id, topic_id)?;
let exists = self
.streams2
- .with_topic_by_id(stream_id, topic_id, |topic| {
- topic.consumer_groups().exists(group_id)
+ .with_topic_root_by_id(stream_id, topic_id, |root| {
+ root.consumer_groups().exists(group_id)
});
if !exists {
return Err(IggyError::ConsumerGroupIdNotFound(0, 0));
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index 9f5dd19c..666e3272 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -18,7 +18,8 @@ use crate::{
personal_access_tokens::personal_access_token::PersonalAccessToken,
polling_consumer::PollingConsumer,
stats::stats::{PartitionStats, StreamStats, TopicStats},
- topics::consumer_group2::Member,
+ streams::stream2,
+ topics::{consumer_group2::Member, topic2},
},
};
@@ -34,8 +35,7 @@ pub enum ShardEvent {
},
CreatedStream2 {
id: usize,
- name: String,
- stats: Arc<StreamStats>,
+ stream: stream2::Stream,
},
CreatedStream {
stream_id: Option<u32>,
@@ -94,14 +94,8 @@ pub enum ShardEvent {
replication_factor: Option<u8>,
},
CreatedTopic2 {
- id: usize,
stream_id: Identifier,
- name: String,
- message_expiry: IggyExpiry,
- compression_algorithm: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- replication_factor: Option<u8>,
- stats: Arc<TopicStats>,
+ topic: topic2::Topic,
},
CreatedConsumerGroup {
stream_id: Identifier,
diff --git a/core/server/src/slab/mod.rs b/core/server/src/slab/mod.rs
index a8185700..840de3d3 100644
--- a/core/server/src/slab/mod.rs
+++ b/core/server/src/slab/mod.rs
@@ -16,4 +16,4 @@ pub trait Keyed {
fn key(&self) -> &Self::Key;
}
-//index: AHashMap<T::Key, usize>,
\ No newline at end of file
+//index: AHashMap<T::Key, usize>,
diff --git a/core/server/src/slab/partitions.rs b/core/server/src/slab/partitions.rs
index 50eb5cc0..8ab5c149 100644
--- a/core/server/src/slab/partitions.rs
+++ b/core/server/src/slab/partitions.rs
@@ -1,113 +1,68 @@
use crate::{
- slab::traits_ext::{
- Borrow, Components, EntityComponentSystem, EntityMarker, IndexComponents, IntoComponents,
- },
+ slab::traits_ext::{Borrow, Delete, EntityComponentSystem, Insert},
streaming::{
- deduplication::message_deduplicator::MessageDeduplicator, partitions::partition2, segments,
+ deduplication::message_deduplicator::MessageDeduplicator,
+ partitions::{
+ consumer_offset,
+ partition2::{self, Partition, PartitionRef},
+ },
+ segments,
stats::stats::PartitionStats,
},
};
use slab::Slab;
-use std::{
- ops::Index,
- sync::{Arc, atomic::AtomicU64},
-};
+use std::sync::{Arc, atomic::AtomicU64};
// TODO: This could be upper limit of partitions per topic, use that value to validate instead of whathever this thing is in `common` crate.
pub const PARTITIONS_CAPACITY: usize = 16384;
-type Id = usize;
+pub type ContainerId = usize;
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub struct Partitions {
root: Slab<partition2::PartitionRoot>,
stats: Slab<Arc<PartitionStats>>,
segments: Slab<Vec<segments::Segment2>>,
message_deduplicator: Slab<Option<MessageDeduplicator>>,
offset: Slab<Arc<AtomicU64>>,
-}
-pub struct Partition {
- root: partition2::PartitionRoot,
- stats: Arc<PartitionStats>,
- message_deduplicator: Option<MessageDeduplicator>,
- offset: Arc<AtomicU64>,
+ consumer_offset: Slab<Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>>,
+ consumer_group_offset: Slab<Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>>,
}
-impl EntityMarker for Partition {}
-
-impl IntoComponents for Partition {
- type Components = (
- partition2::PartitionRoot,
- Arc<PartitionStats>,
- Option<MessageDeduplicator>,
- Arc<AtomicU64>,
- );
-
- fn into_components(self) -> Self::Components {
- (
- self.root,
- self.stats,
- self.message_deduplicator,
- self.offset,
- )
- }
-}
-
-pub struct PartitionRef<'a> {
- root: &'a Slab<partition2::PartitionRoot>,
- stats: &'a Slab<Arc<PartitionStats>>,
- message_deduplicator: &'a Slab<Option<MessageDeduplicator>>,
- offset: &'a Slab<Arc<AtomicU64>>,
-}
+impl Insert for Partitions {
+ type Idx = ContainerId;
+ type Item = Partition;
-impl<'a> From<&'a Partitions> for PartitionRef<'a> {
- fn from(value: &'a Partitions) -> Self {
- PartitionRef {
- root: &value.root,
- stats: &value.stats,
- message_deduplicator: &value.message_deduplicator,
- offset: &value.offset,
- }
+ fn insert(&mut self, item: Self::Item) -> Self::Idx {
+ todo!()
}
}
-impl<'a> IntoComponents for PartitionRef<'a> {
- type Components = (
- &'a Slab<partition2::PartitionRoot>,
- &'a Slab<Arc<PartitionStats>>,
- &'a Slab<Option<MessageDeduplicator>>,
- &'a Slab<Arc<AtomicU64>>,
- );
-
- fn into_components(self) -> Self::Components {
- (
- self.root,
- self.stats,
- self.message_deduplicator,
- self.offset,
- )
+impl Delete for Partitions {
+ type Idx = ContainerId;
+ type Item = Partition;
+
+ fn delete(&mut self, id: Self::Idx) -> Self::Item {
+ todo!()
}
}
-impl<'a> IndexComponents<Id> for PartitionRef<'a> {
- type Output<'b> = (
- &'b partition2::PartitionRoot,
- &'b Arc<PartitionStats>,
- &'b Option<MessageDeduplicator>,
- &'b Arc<AtomicU64>,
- );
-
- fn index(&self, index: Id) -> Self::Output<'_> {
- (
- &self.root[index],
- &self.stats[index],
- &self.message_deduplicator[index],
- &self.offset[index],
+//TODO: those from impls could use a macro aswell.
+impl<'a> From<&'a Partitions> for PartitionRef<'a> {
+ fn from(value: &'a Partitions) -> Self {
+ PartitionRef::new(
+ &value.root,
+ &value.stats,
+ &value.message_deduplicator,
+ &value.offset,
+ &value.consumer_offset,
+ &value.consumer_group_offset,
)
}
}
-impl EntityComponentSystem<Id, Borrow> for Partitions {
+impl EntityComponentSystem<Borrow> for Partitions {
+ type Idx = ContainerId;
type Entity = Partition;
type EntityRef<'a> = PartitionRef<'a>;
@@ -134,11 +89,17 @@ impl Default for Partitions {
segments: Slab::with_capacity(PARTITIONS_CAPACITY),
message_deduplicator: Slab::with_capacity(PARTITIONS_CAPACITY),
offset: Slab::with_capacity(PARTITIONS_CAPACITY),
+ consumer_offset: Slab::with_capacity(PARTITIONS_CAPACITY),
+ consumer_group_offset: Slab::with_capacity(PARTITIONS_CAPACITY),
}
}
}
impl Partitions {
+ pub fn len(&self) -> usize {
+ self.root.len()
+ }
+
pub fn with_stats<T>(&self, f: impl FnOnce(&Slab<Arc<PartitionStats>>) -> T) -> T {
let stats = &self.stats;
f(stats)
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 56bce63d..cc978001 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -2,27 +2,101 @@ use ahash::AHashMap;
use iggy_common::Identifier;
use slab::Slab;
use std::{cell::RefCell, sync::Arc};
+use tokio_rustls::StartHandshake;
use crate::{
- slab::{Keyed, partitions::Partitions, topics::Topics},
- streaming::{
- partitions::partition2, stats::stats::StreamStats, streams::stream2, topics::topic2,
+ shard::stats,
+ slab::{
+ Keyed,
+ partitions::Partitions,
+ topics::Topics,
+ traits_ext::{
+ Delete, DeleteCell, EntityComponentSystem, EntityComponentSystemMutCell, Insert,
+ InsertCell, InteriorMutability, IntoComponents,
+ },
},
+ streaming::{stats::stats::StreamStats, streams::stream2, topics::topic2},
};
const CAPACITY: usize = 1024;
+pub type ContainerId = usize;
pub struct Streams {
- index: RefCell<AHashMap<<stream2::Stream as Keyed>::Key, usize>>,
- container: RefCell<Slab<stream2::Stream>>,
+ index: RefCell<AHashMap<<stream2::StreamRoot as Keyed>::Key, ContainerId>>,
+ root: RefCell<Slab<stream2::StreamRoot>>,
stats: RefCell<Slab<Arc<StreamStats>>>,
}
+impl<'a> From<&'a Streams> for stream2::StreamRef<'a> {
+ fn from(value: &'a Streams) -> Self {
+ let root = value.root.borrow();
+ let stats = value.stats.borrow();
+ stream2::StreamRef::new(root, stats)
+ }
+}
+
+impl<'a> From<&'a Streams> for stream2::StreamRefMut<'a> {
+ fn from(value: &'a Streams) -> Self {
+ let root = value.root.borrow_mut();
+ let stats = value.stats.borrow_mut();
+ stream2::StreamRefMut::new(root, stats)
+ }
+}
+
+impl InsertCell for Streams {
+ type Idx = ContainerId;
+ type Item = stream2::Stream;
+
+ fn insert(&self, item: Self::Item) -> Self::Idx {
+ todo!()
+ }
+}
+
+impl DeleteCell for Streams {
+ type Idx = ContainerId;
+ type Item = stream2::Stream;
+
+ fn delete(&self, id: Self::Idx) -> Self::Item {
+ todo!()
+ }
+}
+
+impl EntityComponentSystem<InteriorMutability> for Streams {
+ type Idx = ContainerId;
+ type Entity = stream2::Stream;
+ type EntityRef<'a> = stream2::StreamRef<'a>;
+
+ fn with<O, F>(&self, f: F) -> O
+ where
+ F: for<'a> FnOnce(Self::EntityRef<'a>) -> O,
+ {
+ f(self.into())
+ }
+
+ async fn with_async<O, F>(&self, f: F) -> O
+ where
+ F: for<'a> AsyncFnOnce(Self::EntityRef<'a>) -> O,
+ {
+ f(self.into()).await
+ }
+}
+
+impl EntityComponentSystemMutCell for Streams {
+ type EntityRefMut<'a> = stream2::StreamRefMut<'a>;
+
+ fn with_mut<O, F>(&self, f: F) -> O
+ where
+ F: for<'a> FnOnce(Self::EntityRefMut<'a>) -> O,
+ {
+ f(self.into())
+ }
+}
+
impl Streams {
pub fn init() -> Self {
Self {
index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
- container: RefCell::new(Slab::with_capacity(CAPACITY)),
+ root: RefCell::new(Slab::with_capacity(CAPACITY)),
stats: RefCell::new(Slab::with_capacity(CAPACITY)),
}
}
@@ -31,7 +105,7 @@ impl Streams {
match id.kind {
iggy_common::IdKind::Numeric => {
let id = id.get_u32_value().unwrap() as usize;
- self.container.borrow().contains(id)
+ self.root.borrow().contains(id)
}
iggy_common::IdKind::String => {
let key = id.get_string_value().unwrap();
@@ -40,7 +114,7 @@ impl Streams {
}
}
- fn get_index(&self, id: &Identifier) -> usize {
+ pub fn get_index(&self, id: &Identifier) -> usize {
match id.kind {
iggy_common::IdKind::Numeric => id.get_u32_value().unwrap() as usize,
iggy_common::IdKind::String => {
@@ -50,89 +124,79 @@ impl Streams {
}
}
- pub fn with_stats<T>(&self, f: impl FnOnce(&Slab<Arc<StreamStats>>) -> T) -> T {
- let stats = self.stats.borrow();
- f(&stats)
- }
-
- pub fn with_stats_by_id<T>(
+ pub fn with_root_by_id<T>(
&self,
id: &Identifier,
- f: impl FnOnce(&Arc<StreamStats>) -> T,
+ f: impl FnOnce(&stream2::StreamRoot) -> T,
) -> T {
- let stream_id = self.with_stream_by_id(id, |stream| stream.id());
- self.with_stats(|stats| {
- let stats = &stats[stream_id];
- f(stats)
- })
- }
-
- pub fn with_stats_mut<T>(&self, f: impl FnOnce(&mut Slab<Arc<StreamStats>>) -> T) -> T {
- let mut stats = self.stats.borrow_mut();
- f(&mut stats)
- }
-
- pub async fn with_async<T>(&self, f: impl AsyncFnOnce(&Slab<stream2::Stream>) -> T) -> T {
- let container = self.container.borrow();
- f(&container).await
+ let id = self.get_index(id);
+ self.with_by_id(id, |(root, _)| f(&root))
}
- pub fn with_index<T>(
+ pub async fn with_root_by_id_async<T>(
&self,
- f: impl FnOnce(&AHashMap<<stream2::Stream as Keyed>::Key, usize>) -> T,
+ id: &Identifier,
+ f: impl AsyncFnOnce(&stream2::StreamRoot) -> T,
) -> T {
- let index = self.index.borrow();
- f(&index)
+ let id = self.get_index(id);
+ self.with_by_id_async(id, async |(root, _)| f(&root).await)
+ .await
}
- pub fn with_index_mut<T>(
+ pub fn with_root_by_id_mut<T>(
&self,
- f: impl FnOnce(&mut AHashMap<<stream2::Stream as Keyed>::Key, usize>) -> T,
+ id: &Identifier,
+ f: impl FnOnce(&mut stream2::StreamRoot) -> T,
) -> T {
- let mut index = self.index.borrow_mut();
- f(&mut index)
- }
-
- pub fn with<T>(&self, f: impl FnOnce(&Slab<stream2::Stream>) -> T) -> T {
- let container = self.container.borrow();
- f(&container)
+ let id = self.get_index(id);
+ self.with_by_id_mut(id, |(mut root, _)| f(&mut root))
}
- pub fn with_mut<T>(&self, f: impl FnOnce(&mut Slab<stream2::Stream>) -> T) -> T {
- let mut container = self.container.borrow_mut();
- f(&mut container)
+ pub fn with_stats<T>(&self, f: impl FnOnce(&Slab<Arc<StreamStats>>) -> T) -> T {
+ self.with(|components| {
+ let (_, stats) = components.into_components();
+ f(&stats)
+ })
}
- pub fn with_stream_by_id<T>(
+ pub fn with_stats_by_id<T>(
&self,
id: &Identifier,
- f: impl FnOnce(&stream2::Stream) -> T,
+ f: impl FnOnce(&Arc<StreamStats>) -> T,
) -> T {
let id = self.get_index(id);
- self.with(|streams| streams[id].invoke(f))
+ 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 async fn with_stream_by_id_async<T>(
+ pub fn with_index<T>(
&self,
- id: &Identifier,
- f: impl AsyncFnOnce(&stream2::Stream) -> T,
+ f: impl FnOnce(&AHashMap<<stream2::StreamRoot as Keyed>::Key, usize>) -> T,
) -> T {
- let id = self.get_index(id);
- self.with_async(async |streams| streams[id].invoke_async(f).await)
- .await
+ let index = self.index.borrow();
+ f(&index)
}
- pub fn with_stream_by_id_mut<T>(
+ pub fn with_index_mut<T>(
&self,
- id: &Identifier,
- f: impl FnOnce(&mut stream2::Stream) -> T,
+ f: impl FnOnce(&mut AHashMap<<stream2::StreamRoot as Keyed>::Key, usize>) -> T,
) -> T {
- let id = self.get_index(id);
- self.with_mut(|streams| streams[id].invoke_mut(f))
+ let mut index = self.index.borrow_mut();
+ f(&mut index)
}
pub fn with_topics<T>(&self, stream_id: &Identifier, f: impl FnOnce(&Topics) -> T) -> T {
- self.with_stream_by_id(stream_id, |stream| f(stream.topics()))
+ let id = self.get_index(stream_id);
+ self.with_by_id(id, |(root, _)| f(root.topics()))
}
pub async fn with_topics_async<T>(
@@ -140,7 +204,8 @@ impl Streams {
stream_id: &Identifier,
f: impl AsyncFnOnce(&Topics) -> T,
) -> T {
- self.with_stream_by_id_async(stream_id, async |stream| f(stream.topics()).await)
+ let id = self.get_index(stream_id);
+ self.with_by_id_async(id, async |(root, _)| f(root.topics()).await)
.await
}
@@ -149,37 +214,48 @@ impl Streams {
stream_id: &Identifier,
f: impl FnOnce(&mut Topics) -> T,
) -> T {
- self.with_stream_by_id_mut(stream_id, |stream| f(stream.topics_mut()))
+ let id = self.get_index(stream_id);
+ self.with_by_id_mut(id, |(mut root, _)| {
+ let topics = root.topics_mut();
+ f(topics)
+ })
}
- pub fn with_topic_by_id<T>(
+ pub fn with_topic_root_by_id<T>(
&self,
stream_id: &Identifier,
id: &Identifier,
- f: impl FnOnce(&topic2::Topic) -> T,
+ f: impl FnOnce(&topic2::TopicRoot) -> T,
) -> T {
- self.with_topics(stream_id, |topics| topics.with_topic_by_id(id, f))
+ self.with_topics(stream_id, |topics| {
+ topics.with_root_by_id(id, |root| f(root))
+ })
}
- pub async fn with_topic_by_id_async<T>(
+ pub async fn with_topic_root_by_id_async<T>(
&self,
stream_id: &Identifier,
id: &Identifier,
- f: impl AsyncFnOnce(&topic2::Topic) -> T,
+ f: impl AsyncFnOnce(&topic2::TopicRoot) -> T,
) -> T {
self.with_topics_async(stream_id, async |topics| {
- topics.with_topic_by_id_async(id, f).await
+ let id = topics.get_index(id);
+ topics
+ .with_by_id_async(id, async |(root, _)| f(&root).await)
+ .await
})
.await
}
- pub fn with_topic_by_id_mut<T>(
+ pub fn with_topic_root_by_id_mut<T>(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
- f: impl FnOnce(&mut topic2::Topic) -> T,
+ f: impl FnOnce(&mut topic2::TopicRoot) -> T,
) -> T {
- self.with_topics_mut(stream_id, |topics| topics.with_topic_by_id_mut(topic_id, f))
+ self.with_topics_mut(stream_id, |topics| {
+ topics.with_root_by_id_mut(topic_id, |root| f(root))
+ })
}
pub fn with_partitions(
@@ -193,6 +269,17 @@ impl Streams {
});
}
+ pub fn with_partitions_mut<T>(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ f: impl FnOnce(&mut Partitions) -> T,
+ ) -> T {
+ self.with_topics_mut(stream_id, |topics| {
+ topics.with_partitions_mut(topic_id, f)
+ })
+ }
+
pub async fn with_partitions_async<T>(
&self,
stream_id: &Identifier,
@@ -206,6 +293,6 @@ impl Streams {
}
pub fn len(&self) -> usize {
- self.container.borrow().len()
+ self.root.borrow().len()
}
}
diff --git a/core/server/src/slab/topics.rs b/core/server/src/slab/topics.rs
index 1c124f78..23198f14 100644
--- a/core/server/src/slab/topics.rs
+++ b/core/server/src/slab/topics.rs
@@ -4,41 +4,116 @@ use slab::Slab;
use std::{cell::RefCell, sync::Arc};
use crate::{
- slab::{partitions::Partitions, Keyed},
- streaming::{partitions::partition2, stats::stats::TopicStats, topics::topic2::{self, TopicRef}},
+ slab::{
+ Keyed,
+ partitions::Partitions,
+ traits_ext::{
+ Delete, DeleteCell, EntityComponentSystem, EntityComponentSystemMutCell, Insert,
+ InsertCell, InteriorMutability, IntoComponents,
+ },
+ },
+ streaming::{
+ partitions::partition2,
+ stats::stats::TopicStats,
+ topics::topic2::{self, TopicRef},
+ },
};
const CAPACITY: usize = 1024;
-pub type SlabId = usize;
+pub type ContainerId = usize;
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub struct Topics {
- index: RefCell<AHashMap<<topic2::TopicRoot as Keyed>::Key, usize>>,
+ index: RefCell<AHashMap<<topic2::TopicRoot as Keyed>::Key, ContainerId>>,
root: RefCell<Slab<topic2::TopicRoot>>,
stats: RefCell<Slab<Arc<TopicStats>>>,
}
+impl InsertCell for Topics {
+ type Idx = ContainerId;
+ type Item = topic2::Topic;
+
+ fn insert(&self, item: Self::Item) -> Self::Idx {
+ todo!()
+ }
+}
+
+impl DeleteCell for Topics {
+ type Idx = ContainerId;
+ type Item = topic2::Topic;
+
+ fn delete(&self, id: Self::Idx) -> Self::Item {
+ todo!()
+ }
+}
+
+//TODO: those from impls could use a macro aswell.
impl<'a> From<&'a Topics> for topic2::TopicRef<'a> {
fn from(value: &'a Topics) -> Self {
- let xd = value.root.borrow();
- TopicRef::new()
+ let root = value.root.borrow();
+ let stats = value.stats.borrow();
+ topic2::TopicRef::new(root, stats)
}
}
-
-impl Topics {
- pub fn init() -> Self {
+impl Default for Topics {
+ fn default() -> Self {
Self {
index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
- container: RefCell::new(Slab::with_capacity(CAPACITY)),
+ root: RefCell::new(Slab::with_capacity(CAPACITY)),
stats: RefCell::new(Slab::with_capacity(CAPACITY)),
}
}
+}
+
+impl<'a> From<&'a Topics> for topic2::TopicRefMut<'a> {
+ fn from(value: &'a Topics) -> Self {
+ let root = value.root.borrow_mut();
+ let stats = value.stats.borrow_mut();
+ topic2::TopicRefMut::new(root, stats)
+ }
+}
+
+impl EntityComponentSystem<InteriorMutability> for Topics {
+ type Idx = ContainerId;
+ type Entity = topic2::Topic;
+ type EntityRef<'a> = topic2::TopicRef<'a>;
+
+ fn with<O, F>(&self, f: F) -> O
+ where
+ F: for<'a> FnOnce(Self::EntityRef<'a>) -> O,
+ {
+ f(self.into())
+ }
+
+ async fn with_async<O, F>(&self, f: F) -> O
+ where
+ F: for<'a> AsyncFnOnce(Self::EntityRef<'a>) -> O,
+ {
+ f(self.into()).await
+ }
+}
+
+impl EntityComponentSystemMutCell for Topics {
+ type EntityRefMut<'a> = topic2::TopicRefMut<'a>;
+
+ fn with_mut<O, F>(&self, f: F) -> O
+ where
+ F: for<'a> FnOnce(Self::EntityRefMut<'a>) -> O,
+ {
+ f(self.into())
+ }
+}
+
+impl Topics {
+ pub fn len(&self) -> usize {
+ self.root.borrow().len()
+ }
pub fn exists(&self, id: &Identifier) -> bool {
match id.kind {
iggy_common::IdKind::Numeric => {
let id = id.get_u32_value().unwrap() as usize;
- self.container.borrow().contains(id)
+ self.root.borrow().contains(id)
}
iggy_common::IdKind::String => {
let key = id.get_string_value().unwrap();
@@ -47,7 +122,7 @@ impl Topics {
}
}
- fn get_index(&self, id: &Identifier) -> usize {
+ pub fn get_index(&self, id: &Identifier) -> usize {
match id.kind {
iggy_common::IdKind::Numeric => id.get_u32_value().unwrap() as usize,
iggy_common::IdKind::String => {
@@ -57,15 +132,95 @@ impl Topics {
}
}
- pub fn len(&self) -> usize {
- self.container.borrow().len()
+ pub fn with_index<T>(
+ &self,
+ f: impl FnOnce(&AHashMap<<topic2::TopicRoot as Keyed>::Key, usize>) -> T,
+ ) -> T {
+ let index = self.index.borrow();
+ f(&index)
+ }
+
+ pub fn with_index_mut<T>(
+ &self,
+ f: impl FnOnce(&mut AHashMap<<topic2::TopicRoot as Keyed>::Key, usize>) -> T,
+ ) -> T {
+ let mut index = self.index.borrow_mut();
+ f(&mut index)
+ }
+
+ pub fn with_root_by_id<T>(
+ &self,
+ id: &Identifier,
+ f: impl FnOnce(&topic2::TopicRoot) -> T,
+ ) -> T {
+ let id = self.get_index(id);
+ self.with_by_id(id, |(root, _)| f(&root))
+ }
+
+ pub async fn with_root_by_id_async<T>(
+ &self,
+ id: &Identifier,
+ f: impl AsyncFnOnce(&topic2::TopicRoot) -> T,
+ ) -> T {
+ let id = self.get_index(id);
+ self.with_by_id_async(id, async |(root, _)| f(&root).await)
+ .await
+ }
+
+ pub fn with_root_by_id_mut<T>(
+ &self,
+ id: &Identifier,
+ f: impl FnOnce(&mut topic2::TopicRoot) -> 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)
+ })
+ }
+
+ 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 async fn with_partitions_async<T>(
+ &self,
+ 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)) {
- self.with_topic_by_id(topic_id, |topic| f(topic.partitions()));
+ let id = self.get_index(topic_id);
+ self.with_by_id(id, |(root, _)| {
+ let partitions = root.partitions();
+ f(partitions)
+ })
}
- pub fn with_partitions_mut(&self, topic_id: &Identifier, f: impl FnOnce(&mut Partitions)) {
- self.with_topic_by_id_mut(topic_id, |topic| f(topic.partitions_mut()));
+ pub fn with_partitions_mut<T>(&self, 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)
+ })
}
}
diff --git a/core/server/src/slab/traits_ext.rs b/core/server/src/slab/traits_ext.rs
index 4323f146..4b860f05 100644
--- a/core/server/src/slab/traits_ext.rs
+++ b/core/server/src/slab/traits_ext.rs
@@ -3,31 +3,51 @@ pub trait IntoComponents {
fn into_components(self) -> Self::Components;
}
-// Marker trait for the entity type.
-pub trait EntityMarker {}
+/// Marker trait for the `Entity`.
+pub trait EntityMarker {
+ type Idx;
+ fn id(&self) -> Self::Idx;
+ fn update_id(&mut self, id: Self::Idx);
+}
+
+/// Insert trait for inserting an `Entity`` into container.
+pub trait Insert {
+ type Idx;
+ type Item: IntoComponents + EntityMarker;
+ fn insert(&mut self, item: Self::Item) -> Self::Idx;
+}
-pub trait Insert<Idx> {
+pub trait InsertCell {
+ type Idx;
type Item: IntoComponents + EntityMarker;
- fn insert(&mut self, item: Self::Item) -> Idx;
+ fn insert(&self, item: Self::Item) -> Self::Idx;
}
-pub trait Delete<Idx> {
+/// Delete trait for deleting an `Entity` from container.
+pub trait Delete {
+ type Idx;
type Item: IntoComponents + EntityMarker;
- fn delete(&mut self, id: Idx) -> Self::Item;
+ fn delete(&mut self, id: Self::Idx) -> Self::Item;
}
-pub trait DeleteCell<Idx> {
+/// Delete trait for deleting an `Entity` from container for container types that use interior mutability.
+pub trait DeleteCell {
+ type Idx;
type Item: IntoComponents + EntityMarker;
- fn delete(&self, id: Idx) -> Self::Item;
+ fn delete(&self, id: Self::Idx) -> Self::Item;
}
-pub trait IndexComponents<Idx: ?Sized> {
- type Output<'a> where Self: 'a;
- fn index(&self, index: Idx) -> Self::Output<'_>;
+/// Trait for getting components by EntityId.
+pub trait IntoComponentsById {
+ type Idx;
+ type Output;
+ fn into_components_by_id(self, index: Self::Idx) -> Self::Output;
}
+/// Marker type for borrow component containers.
pub struct Borrow;
-pub struct RefCell;
+/// Marker type for component containers that use interior mutability.
+pub struct InteriorMutability;
mod private {
pub trait Sealed {}
@@ -35,43 +55,45 @@ mod private {
//TODO: Maybe two seperate traits for Ref and RefMut.
pub trait ComponentsMapping<T>: private::Sealed {
- type Ref<'a> where Self: 'a;
- type RefMut<'a> where Self: 'a;
+ type Ref<'a>;
+ type RefMut<'a>;
}
pub trait ComponentsByIdMapping<T>: private::Sealed {
- // TODO: We will need this to contrain the `EntityRef` and `EntityRefMut` types, so after decomposing they have proper mapping.
- // Similar mechanism to trait from above, but for (T1, T2) -> (&T1, &T2) mapping rather than (T1, T2) -> (&Slab<T1>, &Slab<T2>).
- type Ref<'a> where Self: 'a;
- type RefMut<'a> where Self: 'a;
+ type Ref<'a>;
+ type RefMut<'a>;
}
-macro_rules! impl_components_for_slab_as_refs {
+macro_rules! impl_components_mapping_for_slab {
($T:ident) => {
impl<$T> private::Sealed for ($T,) {}
impl<$T> ComponentsMapping<Borrow> for ($T,)
+ where for<'a> $T :'a
{
- type Ref<'a> = (&'a ::slab::Slab<$T>,) where Self: 'a;
- type RefMut<'a> = (&'a mut ::slab::Slab<$T>,) where Self: 'a;
+ type Ref<'a> = (&'a ::slab::Slab<$T>,);
+ type RefMut<'a> = (&'a mut ::slab::Slab<$T>,);
}
- impl<$T> ComponentsMapping<RefCell> for ($T,)
+ impl<$T> ComponentsMapping<InteriorMutability> for ($T,)
+ where for<'a> $T :'a
{
- type Ref<'a> = (::std::cell::Ref<'a, ::slab::Slab<$T>>,) where Self: 'a;
- type RefMut<'a> = (::std::cell::RefMut<'a, ::slab::Slab<$T>>,) where Self: 'a;
+ type Ref<'a> = (::std::cell::Ref<'a, ::slab::Slab<$T>>,);
+ type RefMut<'a> = (::std::cell::RefMut<'a, ::slab::Slab<$T>>,);
}
impl<$T> ComponentsByIdMapping<Borrow> for ($T,)
+ where for<'a> $T :'a
{
- type Ref<'a> = (&'a $T,) where Self: 'a;
- type RefMut<'a> = (&'a mut $T,) where Self: 'a;
+ type Ref<'a> = (&'a $T,);
+ type RefMut<'a> = (&'a mut $T,);
}
- impl<$T> ComponentsByIdMapping<RefCell> for ($T,)
+ impl<$T> ComponentsByIdMapping<InteriorMutability> for ($T,)
+ where for<'a> $T :'a
{
- type Ref<'a> = (::std::cell::Ref<'a, $T>,) where Self: 'a;
- type RefMut<'a> = (::std::cell::RefMut<'a, $T>,) where Self: 'a;
+ type Ref<'a> = (::std::cell::Ref<'a, $T>,);
+ type RefMut<'a> = (::std::cell::RefMut<'a, $T>,);
}
};
@@ -79,36 +101,44 @@ macro_rules! impl_components_for_slab_as_refs {
impl<$T, $($rest),+> private::Sealed for ($T, $($rest),+) {}
impl<$T, $($rest),+> ComponentsMapping<Borrow> for ($T, $($rest),+)
- where
+ where
+ for<'a> $T :'a,
+ $(for<'a> $rest: 'a),+
{
- type Ref<'a> = (&'a ::slab::Slab<$T>, $(&'a ::slab::Slab<$rest>),+) where Self: 'a;
- type RefMut<'a> = (&'a mut ::slab::Slab<$T>, $(&'a mut ::slab::Slab<$rest>),+) where Self: 'a;
+ type Ref<'a> = (&'a ::slab::Slab<$T>, $(&'a ::slab::Slab<$rest>),+);
+ type RefMut<'a> = (&'a mut ::slab::Slab<$T>, $(&'a mut ::slab::Slab<$rest>),+);
}
- impl<$T, $($rest),+> ComponentsMapping<RefCell> for ($T, $($rest),+)
- where
+ impl<$T, $($rest),+> ComponentsMapping<InteriorMutability> for ($T, $($rest),+)
+ where
+ for<'a> $T :'a,
+ $(for<'a> $rest: 'a),+
{
- type Ref<'a> = (std::cell::Ref<'a, ::slab::Slab<$T>>, $(::std::cell::Ref<'a, ::slab::Slab<$rest>>),+) where Self: 'a;
- type RefMut<'a> = (std::cell::RefMut<'a, ::slab::Slab<$T>>, $(::std::cell::RefMut<'a, ::slab::Slab<$rest>>),+) where Self: 'a;
+ type Ref<'a> = (std::cell::Ref<'a, ::slab::Slab<$T>>, $(::std::cell::Ref<'a, ::slab::Slab<$rest>>),+);
+ type RefMut<'a> = (std::cell::RefMut<'a, ::slab::Slab<$T>>, $(::std::cell::RefMut<'a, ::slab::Slab<$rest>>),+);
}
impl<$T, $($rest),+> ComponentsByIdMapping<Borrow> for ($T, $($rest),+)
- where
+ where
+ for<'a> $T :'a,
+ $(for<'a> $rest: 'a),+
{
- type Ref<'a> = (&'a $T, $(&'a $rest),+) where Self: 'a;
- type RefMut<'a> = (&'a mut $T, $(&'a mut $rest),+) where Self: 'a;
+ type Ref<'a> = (&'a $T, $(&'a $rest),+);
+ type RefMut<'a> = (&'a mut $T, $(&'a mut $rest),+);
}
- impl<$T, $($rest),+> ComponentsByIdMapping<RefCell> for ($T, $($rest),+)
- where
+ impl<$T, $($rest),+> ComponentsByIdMapping<InteriorMutability> for ($T, $($rest),+)
+ where
+ for<'a> $T :'a,
+ $(for<'a> $rest: 'a),+
{
- type Ref<'a> = (std::cell::Ref<'a, $T>, $(::std::cell::Ref<'a, $rest>),+) where Self: 'a;
- type RefMut<'a> = (std::cell::RefMut<'a, $T>, $(::std::cell::RefMut<'a, $rest>),+) where Self: 'a;
+ type Ref<'a> = (std::cell::Ref<'a, $T>, $(::std::cell::Ref<'a, $rest>),+);
+ type RefMut<'a> = (std::cell::RefMut<'a, $T>, $(::std::cell::RefMut<'a, $rest>),+);
}
- impl_components_for_slab_as_refs!($($rest),+);
+ impl_components_mapping_for_slab!($($rest),+);
};
}
-impl_components_for_slab_as_refs!(T1, T2, T3, T4, T5, T6, T7, T8);
+impl_components_mapping_for_slab!(T1, T2, T3, T4, T5, T6, T7, T8);
type Mapping<'a, E, T> = <<E as IntoComponents>::Components as ComponentsMapping<T>>::Ref<'a>;
type MappingMut<'a, E, T> = <<E as IntoComponents>::Components as ComponentsMapping<T>>::RefMut<'a>;
@@ -125,18 +155,16 @@ type MappingByIdMut<'a, E, T> =
// So we lack the ability to immediately discard unnecessary components, which leads to less ergonomic API.
// Damn tradeoffs.
pub type Components<T> = <T as IntoComponents>::Components;
-pub type ComponentsById<'b, Idx, T> = <T as IndexComponents<Idx>>::Output<'b>;
+pub type ComponentsById<'a, T> = <T as IntoComponentsById>::Output;
-pub trait EntityComponentSystem<Idx, T>
+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>>
- + IndexComponents<Idx, Output<'a> = MappingById<'a, Self::Entity, T>> + 'a
- where
- Self::Entity: 'a;
+ type EntityRef<'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
where
@@ -146,56 +174,52 @@ where
where
F: for<'a> AsyncFnOnce(Self::EntityRef<'a>) -> O;
- fn with_by_id<O, F>(&self, id: Idx, f: F) -> O
+ fn with_by_id<O, F>(&self, id: Self::Idx, f: F) -> O
where
- F: for<'a> FnOnce(ComponentsById<'a, Idx, Self::EntityRef<'a>>) -> O,
+ F: for<'a> FnOnce(ComponentsById<'a, Self::EntityRef<'a>>) -> O,
{
- self.with(|components| f(components.index(id)))
+ self.with(|components| f(components.into_components_by_id(id)))
}
- fn with_by_id_async<O, F>(&self, id: Idx, f: F) -> impl Future<Output = O>
+ fn with_by_id_async<O, F>(&self, id: Self::Idx, f: F) -> impl Future<Output = O>
where
- F: for<'a> AsyncFnOnce(ComponentsById<'a, Idx, Self::EntityRef<'a>>) -> O,
+ F: for<'a> AsyncFnOnce(ComponentsById<'a, Self::EntityRef<'a>>) -> O,
{
- self.with_async(async |components| f(components.index(id)).await)
+ self.with_async(async |components| f(components.into_components_by_id(id)).await)
}
}
-/*
-pub trait EntityComponentSystemMut<Idx>: EntityComponentSystem<Idx, Borrow> {
+pub trait EntityComponentSystemMut: EntityComponentSystem<Borrow> {
type EntityRefMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, Borrow>>
- + IndexComponents<Idx, Output<'a> = MappingByIdMut<'a, Self::Entity, Borrow>>
- where
- Self: 'a,
- <Self as EntityComponentSystemMut<Idx>>::EntityRefMut<'a>: 'a;
+ + IntoComponentsById<Idx = Self::Idx, Output = MappingByIdMut<'a, Self::Entity, Borrow>>;
fn with_mut<O, F>(&mut self, f: F) -> O
where
F: for<'a> FnOnce(Self::EntityRefMut<'a>) -> O;
- fn with_by_id_mut<O, F>(&mut self, id: Idx, f: F) -> O
+ fn with_by_id_mut<O, F>(&mut self, id: Self::Idx, f: F) -> O
where
- F: for<'a> FnOnce(ComponentsById<'a, Idx, Self::EntityRefMut<'a>>) -> O,
+ F: for<'a> FnOnce(ComponentsById<'a, Self::EntityRefMut<'a>>) -> O,
{
- self.with_mut(|components| f(components.index(id)))
+ self.with_mut(|components| f(components.into_components_by_id(id)))
}
}
-pub trait EntityComponentSystemMutCell<Idx>: EntityComponentSystem<Idx, RefCell> {
- type EntityRefMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, RefCell>>
- + IndexComponents<Idx, Output<'a> = MappingByIdMut<'a, Self::Entity, RefCell>>
- where
- Self: 'a;
+pub trait EntityComponentSystemMutCell: EntityComponentSystem<InteriorMutability> {
+ type EntityRefMut<'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
where
F: for<'a> FnOnce(Self::EntityRefMut<'a>) -> O;
- fn with_by_id_mut<O, F>(&self, id: Idx, f: F) -> O
+ fn with_by_id_mut<O, F>(&self, id: Self::Idx, f: F) -> O
where
- F: for<'a> FnOnce(ComponentsById<'a, Idx, Self::EntityRefMut<'a>>) -> O,
+ F: for<'a> FnOnce(ComponentsById<'a, Self::EntityRefMut<'a>>) -> O,
{
- self.with_mut(|components| f(components.index(id)))
+ self.with_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 574311d6..c5633048 100644
--- a/core/server/src/streaming/partitions/partition2.rs
+++ b/core/server/src/streaming/partitions/partition2.rs
@@ -1,7 +1,7 @@
use crate::{
slab::{
- partitions::{self, Partitions, SlabId},
- traits_ext::{EntityMarker, IndexComponents, IntoComponents},
+ partitions::{self, Partitions},
+ traits_ext::{EntityMarker, IntoComponents, IntoComponentsById},
},
streaming::{
deduplication::message_deduplicator::{self, MessageDeduplicator},
@@ -41,14 +41,6 @@ impl Partition {
consumer_group_offset,
}
}
-
- pub fn update_id(&mut self, id: usize) {
- self.root.id = id;
- }
-
- pub fn id(&self) -> usize {
- self.root.id
- }
}
impl Clone for Partition {
@@ -64,7 +56,17 @@ impl Clone for Partition {
}
}
-impl EntityMarker for Partition {}
+impl EntityMarker for Partition {
+ type Idx = partitions::ContainerId;
+
+ fn id(&self) -> Self::Idx {
+ self.root.id
+ }
+
+ fn update_id(&mut self, id: Self::Idx) {
+ self.root.id = id;
+ }
+}
impl IntoComponents for Partition {
type Components = (
@@ -113,7 +115,7 @@ impl PartitionRoot {
}
}
-// TODO: Probably move this to the `slab` module
+// TODO: Create a macro to impl those PartitionRef/PartitionRefMut structs and it's traits.
pub struct PartitionRef<'a> {
root: &'a Slab<PartitionRoot>,
stats: &'a Slab<Arc<PartitionStats>>,
@@ -167,7 +169,8 @@ impl<'a> IntoComponents for PartitionRef<'a> {
}
}
-impl<'a> IndexComponents<partitions::SlabId> for PartitionRef<'a> {
+impl<'a> IntoComponentsById for PartitionRef<'a> {
+ type Idx = partitions::ContainerId;
type Output = (
&'a PartitionRoot,
&'a Arc<PartitionStats>,
@@ -177,7 +180,7 @@ impl<'a> IndexComponents<partitions::SlabId> for PartitionRef<'a> {
&'a Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
);
- fn index(&self, index: partitions::SlabId) -> Self::Output {
+ fn into_components_by_id(self, index: Self::Idx) -> Self::Output {
(
&self.root[index],
&self.stats[index],
diff --git a/core/server/src/streaming/segments/segment2.rs b/core/server/src/streaming/segments/segment2.rs
index 9d2f6e70..237c9a3a 100644
--- a/core/server/src/streaming/segments/segment2.rs
+++ b/core/server/src/streaming/segments/segment2.rs
@@ -1,4 +1,4 @@
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub struct Segment2 {
pub id: usize,
}
diff --git a/core/server/src/streaming/streams/stream2.rs b/core/server/src/streaming/streams/stream2.rs
index 6e4eaafb..309e4b95 100644
--- a/core/server/src/streaming/streams/stream2.rs
+++ b/core/server/src/streaming/streams/stream2.rs
@@ -1,29 +1,28 @@
-use bytes::{BufMut, BytesMut};
use iggy_common::IggyTimestamp;
use slab::Slab;
-
-use crate::slab::{Keyed, topics::Topics};
-
-#[derive(Debug)]
-pub struct Stream {
+use std::{
+ cell::{Ref, RefMut},
+ sync::Arc,
+};
+
+use crate::{
+ slab::{
+ Keyed, streams,
+ topics::Topics,
+ traits_ext::{EntityMarker, IntoComponents, IntoComponentsById},
+ },
+ streaming::stats::stats::StreamStats,
+};
+
+#[derive(Debug, Clone)]
+pub struct StreamRoot {
id: usize,
name: String,
created_at: IggyTimestamp,
topics: Topics,
}
-impl Default for Stream {
- fn default() -> Self {
- Self {
- id: 0,
- name: String::new(),
- created_at: IggyTimestamp::now(),
- topics: Topics::init(),
- }
- }
-}
-
-impl Keyed for Stream {
+impl Keyed for StreamRoot {
type Key = String;
fn key(&self) -> &Self::Key {
@@ -31,29 +30,16 @@ impl Keyed for Stream {
}
}
-impl Stream {
- pub fn new(name: String) -> Self {
- let now = IggyTimestamp::now();
+impl StreamRoot {
+ pub fn new(name: String, created_at: IggyTimestamp) -> Self {
Self {
id: 0,
name,
- created_at: now,
- topics: Topics::init(),
+ created_at,
+ topics: Topics::default(),
}
}
- pub fn invoke<T>(&self, f: impl FnOnce(&Self) -> T) -> T {
- f(self)
- }
-
- pub fn invoke_mut<T>(&mut self, f: impl FnOnce(&mut Self) -> T) -> T {
- f(self)
- }
-
- pub async fn invoke_async<T>(&self, f: impl AsyncFnOnce(&Self) -> T) -> T {
- f(self).await
- }
-
pub fn id(&self) -> usize {
self.id
}
@@ -70,13 +56,6 @@ impl Stream {
self.topics.len()
}
- pub fn insert_into(self, container: &mut Slab<Self>) -> usize {
- let idx = container.insert(self);
- let stream = &mut container[idx];
- stream.id = idx;
- idx
- }
-
pub fn topics(&self) -> &Topics {
&self.topics
}
@@ -85,3 +64,110 @@ impl Stream {
&mut self.topics
}
}
+
+#[derive(Debug, Clone)]
+pub struct Stream {
+ root: StreamRoot,
+ stats: Arc<StreamStats>,
+}
+
+impl IntoComponents for Stream {
+ type Components = (StreamRoot, Arc<StreamStats>);
+
+ fn into_components(self) -> Self::Components {
+ (self.root, self.stats)
+ }
+}
+
+impl EntityMarker for Stream {
+ type Idx = streams::ContainerId;
+
+ fn id(&self) -> Self::Idx {
+ self.root.id
+ }
+
+ fn update_id(&mut self, id: Self::Idx) {
+ self.root.id = id;
+ }
+}
+
+impl Stream {
+ pub fn new(name: String, stats: Arc<StreamStats>, created_at: IggyTimestamp) -> Self {
+ let root = StreamRoot::new(name, created_at);
+ Self { root, stats }
+ }
+
+ pub fn stats(&self) -> &Arc<StreamStats> {
+ &self.stats
+ }
+
+ pub fn root(&self) -> &StreamRoot {
+ &self.root
+ }
+}
+
+pub struct StreamRef<'a> {
+ root: Ref<'a, Slab<StreamRoot>>,
+ stats: Ref<'a, Slab<Arc<StreamStats>>>,
+}
+
+impl<'a> StreamRef<'a> {
+ pub fn new(root: Ref<'a, Slab<StreamRoot>>, stats: Ref<'a, Slab<Arc<StreamStats>>>) -> Self {
+ Self { root, stats }
+ }
+}
+
+impl<'a> IntoComponents for StreamRef<'a> {
+ type Components = (Ref<'a, Slab<StreamRoot>>, Ref<'a, Slab<Arc<StreamStats>>>);
+
+ fn into_components(self) -> Self::Components {
+ (self.root, self.stats)
+ }
+}
+
+impl<'a> IntoComponentsById for StreamRef<'a> {
+ type Idx = streams::ContainerId;
+ type Output = (Ref<'a, StreamRoot>, Ref<'a, Arc<StreamStats>>);
+
+ fn into_components_by_id(self, index: Self::Idx) -> Self::Output {
+ let root = Ref::map(self.root, |r| &r[index]);
+ let stats = Ref::map(self.stats, |s| &s[index]);
+ (root, stats)
+ }
+}
+
+pub struct StreamRefMut<'a> {
+ root: RefMut<'a, Slab<StreamRoot>>,
+ stats: RefMut<'a, Slab<Arc<StreamStats>>>,
+}
+
+impl<'a> StreamRefMut<'a> {
+ pub fn new(
+ root: RefMut<'a, Slab<StreamRoot>>,
+ stats: RefMut<'a, Slab<Arc<StreamStats>>>,
+ ) -> Self {
+ Self { root, stats }
+ }
+}
+
+impl<'a> IntoComponents for StreamRefMut<'a> {
+ type Components = (
+ RefMut<'a, Slab<StreamRoot>>,
+ RefMut<'a, Slab<Arc<StreamStats>>>,
+ );
+
+ fn into_components(self) -> Self::Components {
+ (self.root, self.stats)
+ }
+}
+
+impl<'a> IntoComponentsById for StreamRefMut<'a> {
+ type Idx = streams::ContainerId;
+ type Output = (RefMut<'a, StreamRoot>, RefMut<'a, Arc<StreamStats>>);
+
+ fn into_components_by_id(self, index: Self::Idx) -> Self::Output {
+ let root = RefMut::map(self.root, |r| &mut r[index]);
+ let stats = RefMut::map(self.stats, |s| &mut s[index]);
+ (root, stats)
+ }
+}
diff --git a/core/server/src/streaming/topics/topic2.rs b/core/server/src/streaming/topics/topic2.rs
index cc9d1692..01586a87 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -1,21 +1,13 @@
use crate::slab::topics;
-use crate::slab::traits_ext::{EntityMarker, IndexComponents, IntoComponents};
-use crate::state::system::TopicState;
+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 crate::{
- slab::{
- Keyed,
- consumer_groups::ConsumerGroups,
- partitions::{PARTITIONS_CAPACITY, Partitions},
- },
- streaming::partitions::consumer_offset,
-};
use iggy_common::{CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize};
use slab::Slab;
-use std::cell::Ref;
+use std::cell::{Ref, RefMut};
use std::sync::Arc;
-#[derive(Default, Debug)]
+#[derive(Default, Debug, Clone)]
pub struct TopicRoot {
id: usize,
// TODO: This property should be removed, we won't use it in our clustering impl.
@@ -38,7 +30,7 @@ impl Keyed for TopicRoot {
}
}
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub struct Topic {
root: TopicRoot,
stats: Arc<TopicStats>,
@@ -47,27 +39,30 @@ pub struct Topic {
impl Topic {
pub fn new(
name: String,
+ stats: Arc<TopicStats>,
+ created_at: IggyTimestamp,
replication_factor: u8,
message_expiry: IggyExpiry,
compression: CompressionAlgorithm,
max_topic_size: MaxTopicSize,
- stats: Arc<TopicStats>,
) -> Self {
let root = TopicRoot::new(
name,
+ created_at,
replication_factor,
message_expiry,
compression,
max_topic_size,
);
- Self {
- root,
- stats,
- }
+ Self { root, stats }
+ }
+
+ pub fn root(&self) -> &TopicRoot {
+ &self.root
}
}
-impl IntoComponents for Topic{
+impl IntoComponents for Topic {
type Components = (TopicRoot, Arc<TopicStats>);
fn into_components(self) -> Self::Components {
@@ -75,8 +70,18 @@ impl IntoComponents for Topic{
}
}
-impl EntityMarker for Topic {}
+impl EntityMarker for Topic {
+ type Idx = topics::ContainerId;
+ fn id(&self) -> Self::Idx {
+ self.root.id
+ }
+
+ fn update_id(&mut self, id: Self::Idx) {
+ self.root.id = id;
+ }
+}
+// TODO: Create a macro to impl those TopicRef/TopicRefMut structs and it's traits.
pub struct TopicRef<'a> {
root: Ref<'a, Slab<TopicRoot>>,
stats: Ref<'a, Slab<Arc<TopicStats>>>,
@@ -89,31 +94,64 @@ impl<'a> TopicRef<'a> {
}
impl<'a> IntoComponents for TopicRef<'a> {
- type Components = (
- Ref<'a, Slab<TopicRoot>>,
- Ref<'a, Slab<Arc<TopicStats>>>,
- );
+ type Components = (Ref<'a, Slab<TopicRoot>>, Ref<'a, Slab<Arc<TopicStats>>>);
fn into_components(self) -> Self::Components {
(self.root, self.stats)
}
}
+impl<'a> IntoComponentsById for TopicRef<'a> {
+ type Idx = topics::ContainerId;
+ type Output = (Ref<'a, TopicRoot>, Ref<'a, Arc<TopicStats>>);
+
+ fn into_components_by_id(self, index: Self::Idx) -> Self::Output {
+ let root = Ref::map(self.root, |r| &r[index]);
+ let stats = Ref::map(self.stats, |s| &s[index]);
+ (root, stats)
+ }
+}
+
+pub struct TopicRefMut<'a> {
+ root: RefMut<'a, Slab<TopicRoot>>,
+ stats: RefMut<'a, Slab<Arc<TopicStats>>>,
+}
+
+impl<'a> TopicRefMut<'a> {
+ pub fn new(
+ root: RefMut<'a, Slab<TopicRoot>>,
+ stats: RefMut<'a, Slab<Arc<TopicStats>>>,
+ ) -> Self {
+ Self { root, stats }
+ }
+}
-impl<'a> IndexComponents<topics::SlabId> for TopicRef<'a> {
- type Output<'t> = (
- &'t TopicRoot,
- &'t Arc<TopicStats>,
+impl<'a> IntoComponents for TopicRefMut<'a> {
+ type Components = (
+ RefMut<'a, Slab<TopicRoot>>,
+ RefMut<'a, Slab<Arc<TopicStats>>>,
);
- fn index(&self, index: topics::SlabId) -> Self::Output<'_> {
- (&self.root[index], &self.stats[index])
+ fn into_components(self) -> Self::Components {
+ (self.root, self.stats)
+ }
+}
+
+impl<'a> IntoComponentsById for TopicRefMut<'a> {
+ type Idx = topics::ContainerId;
+ type Output = (RefMut<'a, TopicRoot>, RefMut<'a, Arc<TopicStats>>);
+
+ fn into_components_by_id(self, index: Self::Idx) -> Self::Output {
+ let root = RefMut::map(self.root, |r| &mut r[index]);
+ let stats = RefMut::map(self.stats, |s| &mut s[index]);
+ (root, stats)
}
}
impl TopicRoot {
pub fn new(
name: String,
+ created_at: IggyTimestamp,
replication_factor: u8,
message_expiry: IggyExpiry,
compression: CompressionAlgorithm,
@@ -206,4 +244,3 @@ impl TopicRoot {
idx
}
}
-