Create WorldObserverManager

minecraft-server/src/world/change.rs

@@ -1,3 +1,5 @@
+use futures::channel::mpsc::Sender;
+
 use crate::prelude::*;
 
 #[derive(Debug, Clone)]
@@ -99,3 +101,168 @@ impl std::ops::AddAssign<EntityChanges> for EntityChanges {
         self.0 |= rhs.0;
     }
 }
+
+struct WorldObserver {
+    sender: MpscSender<EntityChanges>,
+    ticks: bool,
+    blocks: HashSet<ChunkColumnPosition>,
+    entities: HashSet<ChunkColumnPosition>,
+    nearby_blocks: HashSet<ChunkColumnPosition>,
+    specific_entities: HashSet<Eid>,
+}
+
+#[must_use = "The observer must be added to the manager to be used"]
+pub struct WorldSubscriberBuilder {
+    ticks: bool,
+    blocks: Vec<ChunkColumnPosition>,
+    entities: Vec<ChunkColumnPosition>,
+    nearby_blocks: Vec<NearbyBlockSubscription>,
+    specific_entities: Vec<Eid>,
+}
+
+impl WorldSubscriberBuilder {
+    pub fn new() -> WorldSubscriberBuilder {
+        WorldSubscriberBuilder {
+            ticks: false,
+            blocks: Vec::new(),
+            nearby_blocks: Vec::new(),
+            entities: Vec::new(),
+            specific_entities: Vec::new(),
+        }
+    }
+
+    pub fn with_ticks(mut self) -> WorldSubscriberBuilder {
+        self.ticks = true;
+        self
+    }
+
+    pub fn with_blocks_in_chunk(mut self, position: ChunkColumnPosition) -> WorldSubscriberBuilder {
+        self.blocks.push(position);
+        self
+    }
+
+    pub fn with_entities_in_chunk(mut self, position: ChunkColumnPosition) -> WorldSubscriberBuilder {
+        self.entities.push(position);
+        self
+    }
+
+    pub fn with_nearby_blocks(mut self, position: BlockPosition, radius: u8) -> WorldSubscriberBuilder {
+        self.nearby_blocks.push(NearbyBlockSubscription {
+            position,
+            radius,
+        });
+        self
+    }
+
+    pub fn with_entity(mut self, eid: Eid) -> WorldSubscriberBuilder {
+        self.specific_entities.push(eid);
+        self
+    }
+
+    pub async fn finish(self, eid: Eid, observer_manager: &WorldObserverManager) -> MpscReceiver<EntityChanges> {
+        let (sender, receiver) = mpsc_channel(30);
+        observer_manager.add_subscriber(eid, self, sender).await;
+        receiver
+    }
+}
+
+#[derive(Debug, Clone)]
+struct NearbyBlockSubscription {
+    position: BlockPosition,
+    radius: u8,
+}
+
+pub struct WorldObserverManager {
+    observers: RwLock<HashMap<Eid, WorldObserver>>,
+    ticks: RwLock<HashSet<Eid>>,
+    blocks: RwLock<HashMap<ChunkColumnPosition, HashSet<Eid>>>,
+    entities: RwLock<HashMap<ChunkColumnPosition, HashSet<Eid>>>,
+    nearby_blocks: RwLock<HashMap<ChunkColumnPosition, HashMap<Eid, NearbyBlockSubscription>>>,
+    specific_entities: RwLock<HashMap<Eid, HashSet<Eid>>>,
+}
+
+impl WorldObserverManager {
+    async fn add_subscriber(&self, eid: Eid, observer_builder: WorldSubscriberBuilder, sender: MpscSender<EntityChanges>) {
+        let mut entities = self.observers.write().await;
+        if !observer_builder.blocks.is_empty() {
+            let mut blocks = self.blocks.write().await;
+            for column in &observer_builder.blocks {
+                blocks.entry(column.clone()).or_default().insert(eid);
+            }
+        }
+        if !observer_builder.entities.is_empty() {
+            let mut entities = self.blocks.write().await;
+            for column in &observer_builder.entities {
+                entities.entry(column.clone()).or_default().insert(eid);
+            }
+        }
+        let mut observer_nearby_blocks = HashSet::new();
+        if !observer_builder.nearby_blocks.is_empty() {
+            let mut nearby_blocks = self.nearby_blocks.write().await;
+            for nearby_block in &observer_builder.nearby_blocks {
+                let min_column = BlockPosition {
+                    x: nearby_block.position.x.saturating_sub(nearby_block.radius as i32),
+                    z: nearby_block.position.z.saturating_sub(nearby_block.radius as i32),
+                    y: 0,
+                }.chunk_column();
+                let max_column = BlockPosition {
+                    x: nearby_block.position.x.saturating_add(nearby_block.radius as i32),
+                    z: nearby_block.position.z.saturating_add(nearby_block.radius as i32),
+                    y: 0,
+                }.chunk_column();
+                for cx in min_column.cx..=max_column.cx {
+                    for cz in min_column.cz..=max_column.cz {
+                        nearby_blocks.entry(ChunkColumnPosition {cx: cx, cz: cz}).or_default().insert(eid, nearby_block.clone());
+                        observer_nearby_blocks.insert(ChunkColumnPosition {cx: cx, cz: cz});
+                    }
+                }
+            }
+        }
+        if !observer_builder.specific_entities.is_empty() {
+            let mut specific_entities = self.specific_entities.write().await;
+            for entity in &observer_builder.specific_entities {
+                specific_entities.entry(entity.clone()).or_default().insert(eid);
+            }
+        }
+        entities.insert(eid, WorldObserver {
+            sender,
+            ticks: observer_builder.ticks,
+            blocks: observer_builder.blocks.into_iter().collect(),
+            entities: observer_builder.entities.into_iter().collect(),
+            nearby_blocks: observer_nearby_blocks,
+            specific_entities: observer_builder.specific_entities.into_iter().collect(),
+        });
+    }
+
+    pub async fn remove_subscriber(&self, eid: Eid) {
+        let mut entities = self.observers.write().await;
+        let Some(observer) = entities.remove(&eid) else {return};
+        if observer.ticks {
+            self.ticks.write().await.remove(&eid);
+        }
+        if !observer.blocks.is_empty() {
+            let mut block_subscriptions = self.blocks.write().await;
+            for column in observer.blocks {
+                block_subscriptions.get_mut(&column).map(|set| set.remove(&eid));
+            }
+        }
+        if !observer.nearby_blocks.is_empty() {
+            let mut precise_block_subscriptions = self.nearby_blocks.write().await;
+            for column in observer.nearby_blocks {
+                precise_block_subscriptions.get_mut(&column).map(|map| map.remove(&eid));
+            }
+        }
+        if !observer.entities.is_empty() {
+            let mut entity_subscriptions = self.entities.write().await;
+            for column in observer.entities {
+                entity_subscriptions.get_mut(&column).map(|set| set.remove(&eid));
+            }
+        }
+        if !observer.specific_entities.is_empty() {
+            let mut specific_entity_subscriptions = self.specific_entities.write().await;
+            for entity in observer.specific_entities {
+                specific_entity_subscriptions.get_mut(&entity).map(|set| set.remove(&eid));
+            }
+        }
+    }
+}