| 1 | use crate::block::{BlockCell, ClientID, ItemContent, ItemPtr}; |
| 2 | use crate::block_store::BlockStore; |
| 3 | use crate::branch::{Branch, BranchPtr}; |
| 4 | use crate::doc::{DocAddr, Options}; |
| 5 | use crate::error::Error; |
| 6 | use crate::event::SubdocsEvent; |
| 7 | use crate::id_set::DeleteSet; |
| 8 | use crate::slice::ItemSlice; |
| 9 | use crate::types::{Path, PathSegment, TypeRef}; |
| 10 | use crate::update::PendingUpdate; |
| 11 | use crate::updates::encoder::{Encode, Encoder}; |
| 12 | use crate::StateVector; |
| 13 | use crate::{ |
| 14 | Doc, Observer, OffsetKind, Snapshot, TransactionCleanupEvent, TransactionMut, UpdateEvent, |
| 15 | Uuid, ID, |
| 16 | }; |
| 17 | use arc_swap::{ArcSwap, DefaultStrategy, Guard}; |
| 18 | use async_lock::futures::{Read, Write}; |
| 19 | use async_lock::{RwLock, RwLockReadGuard, RwLockWriteGuard}; |
| 20 | use std::borrow::Borrow; |
| 21 | use std::collections::hash_map::Entry; |
| 22 | use std::collections::{HashMap, HashSet}; |
| 23 | use std::ops::Deref; |
| 24 | use std::sync::Arc; |
| 25 | |
| 26 | /// Store is a core element of a document. It contains all of the information, like block store |
| 27 | /// map of root types, pending updates waiting to be applied once a missing update information |
| 28 | /// arrives and all subscribed callbacks. |
| 29 | pub struct Store { |
| 30 | pub(crate) client_id: ClientID, |
| 31 | pub(crate) offset_kind: OffsetKind, |
| 32 | pub(crate) skip_gc: bool, |
| 33 | |
| 34 | /// Root types (a.k.a. top-level types). These types are defined by users at the document level, |
| 35 | /// they have their own unique names and represent core shared types that expose operations |
| 36 | /// which can be called concurrently by remote peers in a conflict-free manner. |
| 37 | pub(crate) types: HashMap<Arc<str>, Box<Branch>>, |
| 38 | |
| 39 | /// A block store of a current document. It represent all blocks (inserted or tombstoned |
| 40 | /// operations) integrated - and therefore visible - into a current document. |
| 41 | pub(crate) blocks: BlockStore, |
| 42 | |
| 43 | /// A pending update. It contains blocks, which are not yet integrated into `blocks`, usually |
| 44 | /// because due to issues in update exchange, there were some missing blocks that need to be |
| 45 | /// integrated first before the data from `pending` can be applied safely. |
| 46 | pub(crate) pending: Option<PendingUpdate>, |
| 47 | |
| 48 | /// A pending delete set. Just like `pending`, it contains deleted ranges of blocks that have |
| 49 | /// not been yet applied due to missing blocks that prevent `pending` update to be integrated |
| 50 | /// into `blocks`. |
| 51 | pub(crate) pending_ds: Option<DeleteSet>, |
| 52 | |
| 53 | pub(crate) subdocs: HashMap<DocAddr, Doc>, |
| 54 | |
| 55 | pub(crate) events: Option<Box<StoreEvents>>, |
| 56 | |
| 57 | /// Pointer to a parent block - present only if a current document is a sub-document of another |
| 58 | /// document. |
| 59 | pub(crate) parent: Option<ItemPtr>, |
| 60 | |
| 61 | /// Dependencies between items and weak links pointing to these items. |
| 62 | pub(crate) linked_by: HashMap<ItemPtr, HashSet<BranchPtr>>, |
| 63 | } |
| 64 | |
| 65 | impl Store { |
| 66 | /// Create a new empty store in context of a given `client_id`. |
| 67 | pub(crate) fn new(options: &Options) -> Self { |
| 68 | Store { |
| 69 | client_id: options.client_id, |
| 70 | offset_kind: options.offset_kind, |
| 71 | skip_gc: options.skip_gc, |
| 72 | types: HashMap::default(), |
| 73 | blocks: BlockStore::default(), |
| 74 | subdocs: HashMap::default(), |
| 75 | linked_by: HashMap::default(), |
| 76 | events: None, |
| 77 | pending: None, |
| 78 | pending_ds: None, |
| 79 | parent: None, |
| 80 | } |
| 81 | } |
| 82 | |
| 83 | /// If there are any missing updates, this method will return a pending update which contains |
| 84 | /// updates waiting for their predecessors to arrive in order to be integrated. |
| 85 | pub fn pending_update(&self) -> Option<&PendingUpdate> { |
| 86 | self.pending.as_ref() |
| 87 | } |
| 88 | |
| 89 | /// Returns a mutable reference to the pending update if it exists. |
| 90 | pub fn pending_update_mut(&mut self) -> Option<&mut PendingUpdate> { |
| 91 | self.pending.as_mut() |
| 92 | } |
| 93 | |
| 94 | /// If there are some delete updates waiting for missing updates to arrive in order to be |
| 95 | /// applied, this method will return them. |
| 96 | pub fn pending_ds(&self) -> Option<&DeleteSet> { |
| 97 | self.pending_ds.as_ref() |
| 98 | } |
| 99 | |
| 100 | /// Returns a mutable reference to the pending delete set if it exists. |
| 101 | pub fn pending_ds_mut(&mut self) -> Option<&mut DeleteSet> { |
| 102 | self.pending_ds.as_mut() |
| 103 | } |
| 104 | |
| 105 | pub fn is_subdoc(&self) -> bool { |
| 106 | self.parent.is_some() |
| 107 | } |
| 108 | |
| 109 | /// Get the latest clock sequence number observed and integrated into a current store client. |
| 110 | /// This is exclusive value meaning it describes a clock value of the beginning of the next |
| 111 | /// block that's about to be inserted. You cannot use that clock value to find any existing |
| 112 | /// block content. |
| 113 | pub fn get_local_state(&self) -> u32 { |
| 114 | self.blocks.get_clock(&self.client_id) |
| 115 | } |
| 116 | |
| 117 | /// Returns a branch reference to a complex type identified by its pointer. Returns `None` if |
| 118 | /// no such type could be found or was ever defined. |
| 119 | pub(crate) fn get_type<K: Borrow<str>>(&self, key: K) -> Option<BranchPtr> { |
| 120 | let ptr = BranchPtr::from(self.types.get(key.borrow())?); |
| 121 | Some(ptr) |
| 122 | } |
| 123 | |
| 124 | /// Returns a branch reference to a complex type identified by its pointer. Returns `None` if |
| 125 | /// no such type could be found or was ever defined. |
| 126 | pub(crate) fn get_or_create_type<K: Into<Arc<str>>>( |
| 127 | &mut self, |
| 128 | key: K, |
| 129 | type_ref: TypeRef, |
| 130 | ) -> BranchPtr { |
| 131 | let key = key.into(); |
| 132 | match self.types.entry(key.clone()) { |
| 133 | Entry::Occupied(e) => { |
| 134 | let mut branch = BranchPtr::from(e.get()); |
| 135 | branch.repair_type_ref(type_ref); |
| 136 | branch |
| 137 | } |
| 138 | Entry::Vacant(e) => { |
| 139 | let mut branch = Branch::new(type_ref); |
| 140 | let mut branch_ref = BranchPtr::from(&mut branch); |
| 141 | branch_ref.name = Some(key); |
| 142 | e.insert(branch); |
| 143 | branch_ref |
| 144 | } |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | /// Encodes all changes from current transaction block store up to a given `snapshot`. |
| 149 | /// This enables to encode state of a document at some specific point in the past. |
| 150 | pub fn encode_state_from_snapshot<E: Encoder>( |
| 151 | &self, |
| 152 | snapshot: &Snapshot, |
| 153 | encoder: &mut E, |
| 154 | ) -> Result<(), Error> { |
| 155 | if !self.skip_gc { |
| 156 | return Err(Error::Gc); |
| 157 | } |
| 158 | self.write_blocks_to(&snapshot.state_map, encoder); |
| 159 | snapshot.delete_set.encode(encoder); |
| 160 | |
| 161 | Ok(()) |
| 162 | } |
| 163 | |
| 164 | pub(crate) fn write_blocks_to<E: Encoder>(&self, sv: &StateVector, encoder: &mut E) { |
| 165 | let local_sv = self.blocks.get_state_vector(); |
| 166 | let mut diff = Vec::with_capacity(sv.len()); |
| 167 | for (&client_id, &clock) in sv.iter() { |
| 168 | if local_sv.contains_client(&client_id) { |
| 169 | diff.push((client_id, clock.min(local_sv.get(&client_id)))); |
| 170 | } |
| 171 | } |
| 172 | // Write items with higher client ids first |
| 173 | // This heavily improves the conflict algorithm. |
| 174 | diff.sort_by(|a, b| b.0.cmp(&a.0)); |
| 175 | |
| 176 | encoder.write_var(diff.len()); |
| 177 | for (client, clock) in diff { |
| 178 | let blocks = self.blocks.get_client(&client).unwrap(); |
| 179 | let clock = clock.min(blocks.clock() + 1); |
| 180 | let last_idx = blocks.find_pivot(clock - 1).unwrap(); |
| 181 | // write # encoded structs |
| 182 | encoder.write_var(last_idx + 1); |
| 183 | encoder.write_client(client); |
| 184 | encoder.write_var(0); |
| 185 | for i in 0..last_idx { |
| 186 | let block = blocks[i].as_slice(); |
| 187 | block.encode(encoder); |
| 188 | } |
| 189 | let last_block = &blocks[last_idx]; |
| 190 | // write first struct with an offset |
| 191 | let mut slice = last_block.as_slice(); |
| 192 | slice.trim_end(slice.clock_end() - (clock - 1)); |
| 193 | slice.encode(encoder); |
| 194 | } |
| 195 | } |
| 196 | |
| 197 | /// Compute a diff to sync with another client. |
| 198 | /// |
| 199 | /// This is the most efficient method to sync with another client by only |
| 200 | /// syncing the differences. |
| 201 | /// |
| 202 | /// The sync protocol in Yrs/js is: |
| 203 | /// * Send StateVector to the other client. |
| 204 | /// * The other client comutes a minimal diff to sync by using the StateVector. |
| 205 | pub fn encode_diff<E: Encoder>(&self, sv: &StateVector, encoder: &mut E) { |
| 206 | //TODO: this could be actually 2 steps: |
| 207 | // 1. create Diff of block store and remote state vector (it can have lifetime of bock store) |
| 208 | // 2. make Diff implement Encode trait and encode it |
| 209 | // this way we can add some extra utility method on top of Diff (like introspection) without need of decoding it. |
| 210 | self.write_blocks_from(sv, encoder); |
| 211 | let delete_set = DeleteSet::from(&self.blocks); |
| 212 | delete_set.encode(encoder); |
| 213 | } |
| 214 | |
| 215 | pub(crate) fn write_blocks_from<E: Encoder>(&self, sv: &StateVector, encoder: &mut E) { |
| 216 | let local_sv = self.blocks.get_state_vector(); |
| 217 | let mut diff = Self::diff_state_vectors(&local_sv, sv); |
| 218 | |
| 219 | // Write items with higher client ids first |
| 220 | // This heavily improves the conflict algorithm. |
| 221 | diff.sort_by(|a, b| b.0.cmp(&a.0)); |
| 222 | |
| 223 | encoder.write_var(diff.len()); |
| 224 | for (client, clock) in diff { |
| 225 | let blocks = self.blocks.get_client(&client).unwrap(); |
| 226 | let clock = clock.max(blocks.get(0).map(|i| i.clock_start()).unwrap_or_default()); // make sure the first id exists |
| 227 | let start = blocks.find_pivot(clock).unwrap(); |
| 228 | // write # encoded structs |
| 229 | encoder.write_var(blocks.len() - start); |
| 230 | encoder.write_client(client); |
| 231 | encoder.write_var(clock); |
| 232 | let first_block = blocks.get(start).unwrap(); |
| 233 | // write first struct with an offset |
| 234 | let offset = clock - first_block.clock_start(); |
| 235 | let mut slice = first_block.as_slice(); |
| 236 | slice.trim_start(offset); |
| 237 | slice.encode(encoder); |
| 238 | for i in (start + 1)..blocks.len() { |
| 239 | let block = &blocks[i]; |
| 240 | block.as_slice().encode(encoder); |
| 241 | } |
| 242 | } |
| 243 | } |
| 244 | |
| 245 | fn diff_state_vectors(local_sv: &StateVector, remote_sv: &StateVector) -> Vec<(ClientID, u32)> { |
| 246 | let mut diff = Vec::new(); |
| 247 | for (client, &remote_clock) in remote_sv.iter() { |
| 248 | let local_clock = local_sv.get(client); |
| 249 | if local_clock > remote_clock { |
| 250 | diff.push((*client, remote_clock)); |
| 251 | } |
| 252 | } |
| 253 | for (client, _) in local_sv.iter() { |
| 254 | if remote_sv.get(client) == 0 { |
| 255 | diff.push((*client, 0)); |
| 256 | } |
| 257 | } |
| 258 | diff |
| 259 | } |
| 260 | |
| 261 | pub fn get_type_from_path(&self, path: &Path) -> Option<BranchPtr> { |
| 262 | let mut i = path.iter(); |
| 263 | if let Some(PathSegment::Key(root_name)) = i.next() { |
| 264 | let mut current = self.get_type(root_name.clone())?; |
| 265 | while let Some(segment) = i.next() { |
| 266 | match segment { |
| 267 | PathSegment::Key(key) => { |
| 268 | let child = current.map.get(key)?; |
| 269 | if let ItemContent::Type(child_branch) = &child.content { |
| 270 | current = BranchPtr::from(child_branch.as_ref()); |
| 271 | } else { |
| 272 | return None; |
| 273 | } |
| 274 | } |
| 275 | PathSegment::Index(index) => { |
| 276 | if let Some((ItemContent::Type(child_branch), _)) = current.get_at(*index) { |
| 277 | current = child_branch.into(); |
| 278 | } else { |
| 279 | return None; |
| 280 | } |
| 281 | } |
| 282 | } |
| 283 | } |
| 284 | Some(current) |
| 285 | } else { |
| 286 | None |
| 287 | } |
| 288 | } |
| 289 | |
| 290 | /// Consumes current block slice view, materializing it into actual block representation equivalent, |
| 291 | /// splitting underlying block along [ItemSlice::start]/[ItemSlice::end] offsets. |
| 292 | /// |
| 293 | /// Returns a block created this way, that represents the boundaries that current [ItemSlice] |
| 294 | /// was representing. |
| 295 | pub(crate) fn materialize(&mut self, mut slice: ItemSlice) -> ItemPtr { |
| 296 | let id = slice.id().clone(); |
| 297 | let blocks = self.blocks.get_client_mut(&id.client).unwrap(); |
| 298 | let mut links = None; |
| 299 | let item = slice.ptr.deref(); |
| 300 | if item.info.is_linked() { |
| 301 | links = self.linked_by.get(&slice.ptr).cloned(); |
| 302 | } |
| 303 | |
| 304 | let mut index = None; |
| 305 | let mut ptr = if slice.adjacent_left() { |
| 306 | slice.ptr |
| 307 | } else { |
| 308 | let mut i = blocks.find_pivot(id.clock).unwrap(); |
| 309 | if let Some(new) = slice.ptr.splice(slice.start, OffsetKind::Utf16) { |
| 310 | if let Some(source) = links.clone() { |
| 311 | let dest = self.linked_by.entry(ItemPtr::from(&new)).or_default(); |
| 312 | dest.extend(source); |
| 313 | } |
| 314 | blocks.insert(i + 1, BlockCell::Block(new)); |
| 315 | i += 1; |
| 316 | //todo: txn merge blocks insert? |
| 317 | index = Some(i); |
| 318 | } |
| 319 | let ptr = blocks[i].as_item().unwrap(); |
| 320 | slice = ItemSlice::new(ptr, 0, slice.end - slice.start); |
| 321 | ptr |
| 322 | }; |
| 323 | |
| 324 | if !slice.adjacent_right() { |
| 325 | // split block on the right side |
| 326 | let i = if let Some(i) = index { |
| 327 | i |
| 328 | } else { |
| 329 | let last_id = slice.last_id(); |
| 330 | blocks.find_pivot(last_id.clock).unwrap() |
| 331 | }; |
| 332 | let new = ptr.splice(slice.len(), OffsetKind::Utf16).unwrap(); |
| 333 | if let Some(source) = links { |
| 334 | let dest = self.linked_by.entry(ItemPtr::from(&new)).or_default(); |
| 335 | dest.extend(source); |
| 336 | } |
| 337 | blocks.insert(i + 1, BlockCell::Block(new)); |
| 338 | //todo: txn merge blocks insert? |
| 339 | } |
| 340 | |
| 341 | ptr |
| 342 | } |
| 343 | |
| 344 | /// Returns a collection of sub documents linked within the structures of this document store. |
| 345 | pub fn subdocs(&self) -> SubdocsIter { |
| 346 | SubdocsIter(self.subdocs.values()) |
| 347 | } |
| 348 | |
| 349 | /// Returns a collection of globally unique identifiers of sub documents linked within |
| 350 | /// the structures of this document store. |
| 351 | pub fn subdoc_guids(&self) -> SubdocGuids { |
| 352 | SubdocGuids(self.subdocs.values()) |
| 353 | } |
| 354 | |
| 355 | pub(crate) fn follow_redone(&self, id: &ID) -> Option<ItemSlice> { |
| 356 | let mut next_id = Some(*id); |
| 357 | let mut slice = None; |
| 358 | while let Some(next) = next_id.as_mut() { |
| 359 | slice = self.blocks.get_item_clean_start(next); |
| 360 | if let Some(slice) = &slice { |
| 361 | next_id = slice.ptr.redone; |
| 362 | } else { |
| 363 | break; |
| 364 | } |
| 365 | } |
| 366 | slice |
| 367 | } |
| 368 | } |
| 369 | |
| 370 | impl Encode for Store { |
| 371 | /// Encodes the document state to a binary format. |
| 372 | /// |
| 373 | /// Document updates are idempotent and commutative. Caveats: |
| 374 | /// * It doesn't matter in which order document updates are applied. |
| 375 | /// * As long as all clients receive the same document updates, all clients |
| 376 | /// end up with the same content. |
| 377 | /// * Even if an update contains known information, the unknown information |
| 378 | /// is extracted and integrated into the document structure. |
| 379 | fn encode<E: Encoder>(&self, encoder: &mut E) { |
| 380 | self.encode_diff(&StateVector::default(), encoder) |
| 381 | } |
| 382 | } |
| 383 | |
| 384 | impl std::fmt::Debug for Store { |
| 385 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 386 | std::fmt::Display::fmt(self, f) |
| 387 | } |
| 388 | } |
| 389 | |
| 390 | impl std::fmt::Display for Store { |
| 391 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 392 | let mut s = f.debug_struct(&self.client_id.to_string()); |
| 393 | if !self.types.is_empty() { |
| 394 | s.field("root types", &self.types); |
| 395 | } |
| 396 | if !self.blocks.is_empty() { |
| 397 | s.field("blocks", &self.blocks); |
| 398 | } |
| 399 | if let Some(pending) = self.pending.as_ref() { |
| 400 | s.field("pending", pending); |
| 401 | } |
| 402 | if let Some(pending_ds) = self.pending_ds.as_ref() { |
| 403 | s.field("pending delete set", pending_ds); |
| 404 | } |
| 405 | |
| 406 | if let Some(parent) = self.parent.as_ref() { |
| 407 | s.field("parent block", parent.id()); |
| 408 | } |
| 409 | if !self.linked_by.is_empty() { |
| 410 | s.field("links", &self.linked_by); |
| 411 | } |
| 412 | s.finish() |
| 413 | } |
| 414 | } |
| 415 | |
| 416 | #[repr(transparent)] |
| 417 | #[derive(Debug, Clone)] |
| 418 | pub(crate) struct DocStore(pub(crate) Arc<StoreInner>); |
| 419 | |
| 420 | impl DocStore { |
| 421 | pub fn new(options: Options, parent: Option<ItemPtr>) -> Self { |
| 422 | let mut store = Store::new(&options); |
| 423 | let options = ArcSwap::new(options.into()); |
| 424 | store.parent = parent; |
| 425 | DocStore(Arc::new(StoreInner { |
| 426 | options, |
| 427 | store: RwLock::new(store), |
| 428 | })) |
| 429 | } |
| 430 | |
| 431 | pub(crate) fn try_read(&self) -> Option<RwLockReadGuard<Store>> { |
| 432 | self.0.store.try_read() |
| 433 | } |
| 434 | |
| 435 | #[cfg(not(target_family = "wasm"))] |
| 436 | pub(crate) fn read_blocking(&self) -> RwLockReadGuard<Store> { |
| 437 | self.0.store.read_blocking() |
| 438 | } |
| 439 | |
| 440 | #[cfg(target_family = "wasm")] |
| 441 | pub(crate) fn read_blocking(&self) -> RwLockReadGuard<Store> { |
| 442 | self.0.store.try_read().unwrap() |
| 443 | } |
| 444 | |
| 445 | pub(crate) fn read_async(&self) -> Read<Store> { |
| 446 | self.0.store.read() |
| 447 | } |
| 448 | |
| 449 | pub(crate) fn try_write(&self) -> Option<RwLockWriteGuard<Store>> { |
| 450 | self.0.store.try_write() |
| 451 | } |
| 452 | |
| 453 | #[cfg(not(target_family = "wasm"))] |
| 454 | pub(crate) fn write_blocking(&self) -> RwLockWriteGuard<Store> { |
| 455 | self.0.store.write_blocking() |
| 456 | } |
| 457 | |
| 458 | #[cfg(target_family = "wasm")] |
| 459 | pub(crate) fn write_blocking(&self) -> RwLockWriteGuard<Store> { |
| 460 | self.0.store.try_write().unwrap() |
| 461 | } |
| 462 | |
| 463 | pub(crate) fn write_async(&self) -> Write<Store> { |
| 464 | self.0.store.write() |
| 465 | } |
| 466 | |
| 467 | pub(crate) fn options(&self) -> Guard<Arc<Options>, DefaultStrategy> { |
| 468 | self.0.options.load() |
| 469 | } |
| 470 | |
| 471 | /// Sets [Doc::should_load] flag, returning previous value. |
| 472 | pub(crate) fn set_should_load(&self, should_load: bool) -> bool { |
| 473 | self.0 |
| 474 | .options |
| 475 | .rcu(|options| { |
| 476 | let mut options = options.deref().clone(); |
| 477 | options.should_load = should_load; |
| 478 | options |
| 479 | }) |
| 480 | .should_load |
| 481 | } |
| 482 | |
| 483 | pub(crate) fn set_subdoc_data(&self, client_id: ClientID, collection_id: Option<Arc<str>>) { |
| 484 | self.0.options.rcu(|options| { |
| 485 | let mut options = options.deref().clone(); |
| 486 | options.client_id = client_id; |
| 487 | if options.collection_id.is_none() { |
| 488 | options.collection_id = collection_id.clone(); |
| 489 | } |
| 490 | options |
| 491 | }); |
| 492 | } |
| 493 | } |
| 494 | |
| 495 | #[derive(Debug)] |
| 496 | pub(crate) struct StoreInner { |
| 497 | options: ArcSwap<Options>, |
| 498 | store: RwLock<Store>, |
| 499 | } |
| 500 | |
| 501 | #[repr(transparent)] |
| 502 | pub struct SubdocsIter<'doc>(std::collections::hash_map::Values<'doc, DocAddr, Doc>); |
| 503 | |
| 504 | impl<'doc> Iterator for SubdocsIter<'doc> { |
| 505 | type Item = &'doc Doc; |
| 506 | |
| 507 | fn next(&mut self) -> Option<Self::Item> { |
| 508 | self.0.next() |
| 509 | } |
| 510 | } |
| 511 | |
| 512 | #[repr(transparent)] |
| 513 | pub struct SubdocGuids<'doc>(std::collections::hash_map::Values<'doc, DocAddr, Doc>); |
| 514 | |
| 515 | impl<'doc> Iterator for SubdocGuids<'doc> { |
| 516 | type Item = Uuid; |
| 517 | |
| 518 | fn next(&mut self) -> Option<Self::Item> { |
| 519 | let d = self.0.next()?; |
| 520 | Some(d.guid()) |
| 521 | } |
| 522 | } |
| 523 | |
| 524 | #[cfg(feature = "sync")] |
| 525 | pub type TransactionCleanupFn = |
| 526 | Box<dyn Fn(&TransactionMut, &TransactionCleanupEvent) + Send + Sync + 'static>; |
| 527 | #[cfg(feature = "sync")] |
| 528 | pub type AfterTransactionFn = Box<dyn Fn(&mut TransactionMut) + Send + Sync + 'static>; |
| 529 | #[cfg(feature = "sync")] |
| 530 | pub type UpdateFn = Box<dyn Fn(&TransactionMut, &UpdateEvent) + Send + Sync + 'static>; |
| 531 | #[cfg(feature = "sync")] |
| 532 | pub type SubdocsFn = Box<dyn Fn(&TransactionMut, &SubdocsEvent) + Send + Sync + 'static>; |
| 533 | #[cfg(feature = "sync")] |
| 534 | pub type DestroyFn = Box<dyn Fn(&TransactionMut, &Doc) + Send + Sync + 'static>; |
| 535 | |
| 536 | #[cfg(not(feature = "sync"))] |
| 537 | pub type TransactionCleanupFn = Box<dyn Fn(&TransactionMut, &TransactionCleanupEvent) + 'static>; |
| 538 | #[cfg(not(feature = "sync"))] |
| 539 | pub type AfterTransactionFn = Box<dyn Fn(&mut TransactionMut) + 'static>; |
| 540 | #[cfg(not(feature = "sync"))] |
| 541 | pub type UpdateFn = Box<dyn Fn(&TransactionMut, &UpdateEvent) + 'static>; |
| 542 | #[cfg(not(feature = "sync"))] |
| 543 | pub type SubdocsFn = Box<dyn Fn(&TransactionMut, &SubdocsEvent) + 'static>; |
| 544 | #[cfg(not(feature = "sync"))] |
| 545 | pub type DestroyFn = Box<dyn Fn(&TransactionMut, &Doc) + 'static>; |
| 546 | |
| 547 | #[derive(Default)] |
| 548 | pub struct StoreEvents { |
| 549 | /// Handles subscriptions for the transaction cleanup event. Events are called with the |
| 550 | /// newest updates once they are committed and compacted. |
| 551 | pub transaction_cleanup_events: Observer<TransactionCleanupFn>, |
| 552 | |
| 553 | /// Handles subscriptions for the `afterTransactionCleanup` event. Events are called with the |
| 554 | /// newest updates once they are committed and compacted. |
| 555 | pub after_transaction_events: Observer<AfterTransactionFn>, |
| 556 | |
| 557 | /// A subscription handler. It contains all callbacks with registered by user functions that |
| 558 | /// are supposed to be called, once a new update arrives. |
| 559 | pub update_v1_events: Observer<UpdateFn>, |
| 560 | |
| 561 | /// A subscription handler. It contains all callbacks with registered by user functions that |
| 562 | /// are supposed to be called, once a new update arrives. |
| 563 | pub update_v2_events: Observer<UpdateFn>, |
| 564 | |
| 565 | /// Handles subscriptions for subdocs events. |
| 566 | pub subdocs_events: Observer<SubdocsFn>, |
| 567 | |
| 568 | pub destroy_events: Observer<DestroyFn>, |
| 569 | } |
| 570 | |
| 571 | impl StoreEvents { |
| 572 | pub fn emit_update_v1(&self, txn: &TransactionMut) { |
| 573 | if self.update_v1_events.has_subscribers() { |
| 574 | if !txn.delete_set.is_empty() || txn.after_state != txn.before_state { |
| 575 | // produce update only if anything changed |
| 576 | let update = UpdateEvent::new_v1(txn); |
| 577 | self.update_v1_events |
| 578 | .trigger(|callback| callback(txn, &update)); |
| 579 | } |
| 580 | } |
| 581 | } |
| 582 | |
| 583 | pub fn emit_update_v2(&self, txn: &TransactionMut) { |
| 584 | if self.update_v2_events.has_subscribers() { |
| 585 | if !txn.delete_set.is_empty() || txn.after_state != txn.before_state { |
| 586 | // produce update only if anything changed |
| 587 | let update = UpdateEvent::new_v2(txn); |
| 588 | self.update_v2_events.trigger(|fun| fun(txn, &update)); |
| 589 | } |
| 590 | } |
| 591 | } |
| 592 | |
| 593 | pub fn emit_after_transaction(&self, txn: &mut TransactionMut) { |
| 594 | self.after_transaction_events.trigger(|fun| fun(txn)); |
| 595 | } |
| 596 | |
| 597 | pub fn emit_transaction_cleanup(&self, txn: &TransactionMut) { |
| 598 | if self.transaction_cleanup_events.has_subscribers() { |
| 599 | let event = TransactionCleanupEvent::new(txn); |
| 600 | self.transaction_cleanup_events |
| 601 | .trigger(|fun| fun(txn, &event)); |
| 602 | } |
| 603 | } |
| 604 | } |
| 605 | |
amadan