| 37 | } |
| 38 | |
| 39 | fn insert(&mut self, def: ChangeStreamDef) { |
| 40 | let key_name = (def.tenant_id, def.name.clone()); |
| 41 | let key_coll = (def.tenant_id, def.collection.clone()); |
| 42 | let def_arc = Arc::new(def); |
| 43 | |
| 44 | // Replace any prior entry under this name — removes its collection-index entry first. |
| 45 | if let Some(prev) = self.by_name.remove(&key_name) { |
| 46 | let prev_coll_key = (prev.tenant_id, prev.collection.clone()); |
| 47 | if let Some(bucket) = self.by_collection.get_mut(&prev_coll_key) { |
| 48 | bucket.retain(|d| d.name != prev.name); |
| 49 | if bucket.is_empty() { |
| 50 | self.by_collection.remove(&prev_coll_key); |
| 51 | } |
| 52 | } |
| 53 | } |
| 54 | |
| 55 | self.by_collection |
| 56 | .entry(key_coll) |
| 57 | .or_default() |
| 58 | .push(Arc::clone(&def_arc)); |
| 59 | self.by_name.insert(key_name, def_arc); |
| 60 | } |
| 61 | |
| 62 | fn remove(&mut self, tenant_id: u64, name: &str) -> bool { |
| 63 | let key_name = (tenant_id, name.to_string()); |