| 1 | use crate::block::{Item, ItemContent, ItemPosition, ItemPtr, Prelim, ID}; |
| 2 | use crate::branch::{Branch, BranchPtr}; |
| 3 | use crate::doc::DocAddr; |
| 4 | use crate::error::{Error, UpdateError}; |
| 5 | use crate::event::SubdocsEvent; |
| 6 | use crate::gc::GCCollector; |
| 7 | use crate::id_set::DeleteSet; |
| 8 | use crate::iter::TxnIterator; |
| 9 | use crate::slice::BlockSlice; |
| 10 | use crate::store::{Store, StoreEvents, SubdocGuids, SubdocsIter}; |
| 11 | use crate::types::{Event, Events, RootRef, TypePtr, TypeRef}; |
| 12 | use crate::update::Update; |
| 13 | use crate::updates::encoder::{Encode, Encoder, EncoderV1, EncoderV2}; |
| 14 | use crate::utils::OptionExt; |
| 15 | use crate::{ |
| 16 | merge_updates_v1, merge_updates_v2, ArrayRef, BranchID, Doc, MapRef, Out, Snapshot, |
| 17 | StateVector, TextRef, Transact, XmlElementRef, XmlFragmentRef, XmlTextRef, |
| 18 | }; |
| 19 | use async_lock::{RwLockReadGuard, RwLockWriteGuard}; |
| 20 | use smallvec::SmallVec; |
| 21 | use std::collections::{HashMap, HashSet, VecDeque}; |
| 22 | use std::fmt::Formatter; |
| 23 | use std::hash::Hash; |
| 24 | use std::ops::{Deref, DerefMut}; |
| 25 | use std::pin::Pin; |
| 26 | use std::sync::Arc; |
| 27 | |
| 28 | /// Trait defining read capabilities present in a transaction. Implemented by both lightweight |
| 29 | /// [read-only](Transaction) and [read-write](TransactionMut) transactions. |
| 30 | pub trait ReadTxn: Sized { |
| 31 | fn store(&self) -> &Store; |
| 32 | |
| 33 | /// Returns state vector describing current state of the updates. |
| 34 | fn state_vector(&self) -> StateVector { |
| 35 | self.store().blocks.get_state_vector() |
| 36 | } |
| 37 | |
| 38 | /// Returns a snapshot which describes a current state of updates and removals made within |
| 39 | /// the corresponding document. |
| 40 | fn snapshot(&self) -> Snapshot { |
| 41 | let store = self.store(); |
| 42 | let blocks = &store.blocks; |
| 43 | let sv = blocks.get_state_vector(); |
| 44 | let ds = DeleteSet::from(blocks); |
| 45 | Snapshot::new(sv, ds) |
| 46 | } |
| 47 | |
| 48 | /// Encodes all changes from current transaction block store up to a given `snapshot`. |
| 49 | /// This enables to encode state of a document at some specific point in the past. |
| 50 | fn encode_state_from_snapshot<E: Encoder>( |
| 51 | &self, |
| 52 | snapshot: &Snapshot, |
| 53 | encoder: &mut E, |
| 54 | ) -> Result<(), Error> { |
| 55 | self.store().encode_state_from_snapshot(snapshot, encoder) |
| 56 | } |
| 57 | |
| 58 | /// Encodes the difference between remote peer state given its `state_vector` and the state |
| 59 | /// of a current local peer. |
| 60 | /// |
| 61 | /// # Differences between alternative methods |
| 62 | /// |
| 63 | /// - [Self::encode_state_as_update] encodes full document state including pending updates and |
| 64 | /// entire delete set. |
| 65 | /// - [Self::encode_diff] encodes only the difference between the current state and |
| 66 | /// the given state vector, including entire delete set. Pending updates are not included. |
| 67 | /// - [TransactionMut::encode_update] encodes only inserts and deletes made within the scope |
| 68 | /// of the current transaction. |
| 69 | fn encode_diff<E: Encoder>(&self, state_vector: &StateVector, encoder: &mut E) { |
| 70 | self.store().encode_diff(state_vector, encoder) |
| 71 | } |
| 72 | |
| 73 | /// Encodes the difference between remote peer state given its `state_vector` and the state |
| 74 | /// of a current local peer, using lib0 v1 encoding. |
| 75 | /// |
| 76 | /// # Differences between alternative methods |
| 77 | /// |
| 78 | /// - [Self::encode_state_as_update_v1] encodes full document state including pending updates |
| 79 | /// and entire delete set. |
| 80 | /// - [Self::encode_diff_v1] encodes only the difference between the current state and |
| 81 | /// the given state vector, including entire delete set. Pending updates are not included. |
| 82 | /// - [TransactionMut::encode_update_v1] encodes only inserts and deletes made within the scope |
| 83 | /// of the current transaction. |
| 84 | fn encode_diff_v1(&self, state_vector: &StateVector) -> Vec<u8> { |
| 85 | let mut encoder = EncoderV1::new(); |
| 86 | self.encode_diff(state_vector, &mut encoder); |
| 87 | encoder.to_vec() |
| 88 | } |
| 89 | |
| 90 | /// Encodes the difference between remote peer state given its `state_vector` and the state |
| 91 | /// of a current local peer, using lib0 v2 encoding. |
| 92 | /// |
| 93 | /// # Differences between alternative methods |
| 94 | /// |
| 95 | /// - [Self::encode_state_as_update_v2] encodes full document state including pending updates |
| 96 | /// and entire delete set. |
| 97 | /// - [Self::encode_diff_v2] encodes only the difference between the current state and |
| 98 | /// the given state vector, including entire delete set. Pending updates are not included. |
| 99 | /// - [TransactionMut::encode_update_v2] encodes only inserts and deletes made within the scope |
| 100 | /// of the current transaction. |
| 101 | fn encode_diff_v2(&self, state_vector: &StateVector) -> Vec<u8> { |
| 102 | let mut encoder = EncoderV2::new(); |
| 103 | self.encode_diff(state_vector, &mut encoder); |
| 104 | encoder.to_vec() |
| 105 | } |
| 106 | |
| 107 | /// Encodes the difference between remote peer state given its `state_vector` and the state |
| 108 | /// of a current local peer. Also includes pending updates which were not yet integrated into |
| 109 | /// the main document state and entire delete set. |
| 110 | /// |
| 111 | /// # Differences between alternative methods |
| 112 | /// |
| 113 | /// - [Self::encode_state_as_update] encodes full document state including pending updates and |
| 114 | /// entire delete set. |
| 115 | /// - [Self::encode_diff] encodes only the difference between the current state and |
| 116 | /// the given state vector, including entire delete set. Pending updates are not included. |
| 117 | /// - [TransactionMut::encode_update] encodes only inserts and deletes made within the scope |
| 118 | /// of the current transaction. |
| 119 | fn encode_state_as_update<E: Encoder>(&self, sv: &StateVector, encoder: &mut E) { |
| 120 | let store = self.store(); |
| 121 | store.write_blocks_from(sv, encoder); |
| 122 | let ds = DeleteSet::from(&store.blocks); |
| 123 | ds.encode(encoder); |
| 124 | } |
| 125 | |
| 126 | /// Encodes the difference between remote peer state given its `state_vector` and the state |
| 127 | /// of a current local peer, using lib0 v1 encoding. Also includes pending updates which were |
| 128 | /// not yet integrated into the main document state and entire delete set. |
| 129 | /// |
| 130 | /// # Differences between alternative methods |
| 131 | /// |
| 132 | /// - [Self::encode_state_as_update_v1] encodes full document state including pending updates |
| 133 | /// and entire delete set. |
| 134 | /// - [Self::encode_diff_v1] encodes only the difference between the current state and |
| 135 | /// the given state vector, including entire delete set. Pending updates are not included. |
| 136 | /// - [TransactionMut::encode_update_v1] encodes only inserts and deletes made within the scope |
| 137 | /// of the current transaction. |
| 138 | fn encode_state_as_update_v1(&self, sv: &StateVector) -> Vec<u8> { |
| 139 | let mut encoder = EncoderV1::new(); |
| 140 | self.encode_state_as_update(sv, &mut encoder); |
| 141 | // check for pending data |
| 142 | merge_pending_v1(encoder.to_vec(), self.store()) |
| 143 | } |
| 144 | |
| 145 | /// Encodes the difference between remote peer state given its `state_vector` and the state |
| 146 | /// of a current local peer, using lib0 v2 encoding. Also includes pending updates which were |
| 147 | /// not yet integrated into the main document state and entire delete set. |
| 148 | /// |
| 149 | /// # Differences between alternative methods |
| 150 | /// |
| 151 | /// - [Self::encode_state_as_update_v2] encodes full document state including pending updates |
| 152 | /// and entire delete set. |
| 153 | /// - [Self::encode_diff_v2] encodes only the difference between the current state and |
| 154 | /// the given state vector, including entire delete set. Pending updates are not included. |
| 155 | /// - [TransactionMut::encode_update_v2] encodes only inserts and deletes made within the scope |
| 156 | /// of the current transaction. |
| 157 | fn encode_state_as_update_v2(&self, sv: &StateVector) -> Vec<u8> { |
| 158 | let mut encoder = EncoderV2::new(); |
| 159 | self.encode_state_as_update(sv, &mut encoder); |
| 160 | |
| 161 | // check for pending data |
| 162 | merge_pending_v2(encoder.to_vec(), self.store()) |
| 163 | } |
| 164 | |
| 165 | /// Returns an iterator over top level (root) shared types available in current [Doc]. |
| 166 | fn root_refs(&self) -> RootRefs { |
| 167 | let store = self.store(); |
| 168 | RootRefs(store.types.iter()) |
| 169 | } |
| 170 | |
| 171 | /// Returns a collection of globally unique identifiers of sub documents linked within |
| 172 | /// the structures of this document store. |
| 173 | fn subdoc_guids(&self) -> SubdocGuids { |
| 174 | let store = self.store(); |
| 175 | store.subdoc_guids() |
| 176 | } |
| 177 | |
| 178 | /// Returns a collection of sub documents linked within the structures of this document store. |
| 179 | fn subdocs(&self) -> SubdocsIter { |
| 180 | let store = self.store(); |
| 181 | store.subdocs() |
| 182 | } |
| 183 | |
| 184 | /// Returns a [TextRef] data structure stored under a given `name`. Text structures are used for |
| 185 | /// collaborative text editing: they expose operations to append and remove chunks of text, |
| 186 | /// which are free to execute concurrently by multiple peers over remote boundaries. |
| 187 | /// |
| 188 | /// If not structure under defined `name` existed before, [None] will be returned. |
| 189 | /// |
| 190 | /// If a structure under defined `name` already existed, but its type was different it will be |
| 191 | /// reinterpreted as a text (in such case a sequence component of complex data type will be |
| 192 | /// interpreted as a list of text chunks). |
| 193 | #[inline] |
| 194 | fn get_text<N: Into<Arc<str>>>(&self, name: N) -> Option<TextRef> { |
| 195 | TextRef::root(name).get(self) |
| 196 | } |
| 197 | |
| 198 | /// Returns an [ArrayRef] data structure stored under a given `name`. Array structures are used for |
| 199 | /// storing a sequences of elements in ordered manner, positioning given element accordingly |
| 200 | /// to its index. |
| 201 | /// |
| 202 | /// If not structure under defined `name` existed before, [None] will be returned. |
| 203 | /// |
| 204 | /// If a structure under defined `name` already existed, but its type was different it will be |
| 205 | /// reinterpreted as an array (in such case a sequence component of complex data type will be |
| 206 | /// interpreted as a list of inserted values). |
| 207 | #[inline] |
| 208 | fn get_array<N: Into<Arc<str>>>(&self, name: N) -> Option<ArrayRef> { |
| 209 | ArrayRef::root(name).get(self) |
| 210 | } |
| 211 | |
| 212 | /// Returns a [MapRef] data structure stored under a given `name`. Maps are used to store key-value |
| 213 | /// pairs associated. These values can be primitive data (similar but not limited to |
| 214 | /// a JavaScript Object Notation) as well as other shared types (Yrs maps, arrays, text |
| 215 | /// structures etc.), enabling to construct a complex recursive tree structures. |
| 216 | /// |
| 217 | /// If not structure under defined `name` existed before, [None] will be returned. |
| 218 | /// |
| 219 | /// If a structure under defined `name` already existed, but its type was different it will be |
| 220 | /// reinterpreted as a map (in such case a map component of complex data type will be |
| 221 | /// interpreted as native map). |
| 222 | #[inline] |
| 223 | fn get_map<N: Into<Arc<str>>>(&self, name: N) -> Option<MapRef> { |
| 224 | MapRef::root(name).get(self) |
| 225 | } |
| 226 | |
| 227 | /// Returns a [XmlFragmentRef] data structure stored under a given `name`. XML elements represent |
| 228 | /// nodes of XML document. They can contain attributes (key-value pairs, both of string type) |
| 229 | /// and other nested XML elements or text values, which are stored in their insertion |
| 230 | /// order. |
| 231 | /// |
| 232 | /// If not structure under defined `name` existed before, [None] will be returned. |
| 233 | /// |
| 234 | /// If a structure under defined `name` already existed, but its type was different it will be |
| 235 | /// reinterpreted as a XML element (in such case a map component of complex data type will be |
| 236 | /// interpreted as map of its attributes, while a sequence component - as a list of its child |
| 237 | /// XML nodes). |
| 238 | #[inline] |
| 239 | fn get_xml_fragment<N: Into<Arc<str>>>(&self, name: N) -> Option<XmlFragmentRef> { |
| 240 | XmlFragmentRef::root(name).get(self) |
| 241 | } |
| 242 | |
| 243 | fn get<S: AsRef<str>>(&self, name: S) -> Option<Out> { |
| 244 | let value = self.store().types.get(name.as_ref())?; |
| 245 | let ptr = BranchPtr::from(&*value); |
| 246 | match &ptr.type_ref { |
| 247 | TypeRef::Array => Some(Out::YArray(ArrayRef::from(ptr))), |
| 248 | TypeRef::Map => Some(Out::YMap(MapRef::from(ptr))), |
| 249 | TypeRef::Text => Some(Out::YText(TextRef::from(ptr))), |
| 250 | TypeRef::XmlElement(_) => Some(Out::YXmlElement(XmlElementRef::from(ptr))), |
| 251 | TypeRef::XmlFragment => Some(Out::YXmlFragment(XmlFragmentRef::from(ptr))), |
| 252 | TypeRef::XmlHook => None, |
| 253 | TypeRef::XmlText => Some(Out::YXmlText(XmlTextRef::from(ptr))), |
| 254 | TypeRef::SubDoc => Some(Out::YDoc(ptr.as_subdoc()?)), |
| 255 | #[cfg(feature = "weak")] |
| 256 | TypeRef::WeakLink(_) => Some(Out::YWeakLink(crate::WeakRef::from(ptr))), |
| 257 | TypeRef::Undefined => Some(Out::UndefinedRef(ptr)), |
| 258 | } |
| 259 | } |
| 260 | |
| 261 | /// If current document has been inserted as a sub-document, returns a reference to a parent |
| 262 | /// document, which contains it. |
| 263 | fn parent_doc(&self) -> Option<Doc> { |
| 264 | if let Some(item) = self.store().parent.as_deref() { |
| 265 | if let ItemContent::Doc(parent_doc, _) = &item.content { |
| 266 | return parent_doc.clone(); |
| 267 | } |
| 268 | } |
| 269 | |
| 270 | None |
| 271 | } |
| 272 | |
| 273 | /// If current document has been inserted as a sub-document, returns its [BranchID]. |
| 274 | fn branch_id(&self) -> Option<BranchID> { |
| 275 | if let Some(item) = self.store().parent { |
| 276 | Some(BranchID::Nested(item.id)) |
| 277 | } else { |
| 278 | None |
| 279 | } |
| 280 | } |
| 281 | |
| 282 | /// Returns `true` if current document has any pending updates that are not yet |
| 283 | /// integrated into the document. |
| 284 | fn has_missing_updates(&self) -> bool { |
| 285 | let store = self.store(); |
| 286 | store.pending.is_some() || store.pending_ds.is_some() |
| 287 | } |
| 288 | } |
| 289 | |
| 290 | pub trait WriteTxn: Sized { |
| 291 | fn store_mut(&mut self) -> &mut Store; |
| 292 | fn subdocs_mut(&mut self) -> &mut Subdocs; |
| 293 | |
| 294 | /// Returns a [TextRef] data structure stored under a given `name`. Text structures are used for |
| 295 | /// collaborative text editing: they expose operations to append and remove chunks of text, |
| 296 | /// which are free to execute concurrently by multiple peers over remote boundaries. |
| 297 | /// |
| 298 | /// If no structure under defined `name` existed before, it will be created and returned |
| 299 | /// instead. |
| 300 | /// |
| 301 | /// If a structure under defined `name` already existed, but its type was different it will be |
| 302 | /// reinterpreted as a text (in such case a sequence component of complex data type will be |
| 303 | /// interpreted as a list of text chunks). |
| 304 | fn get_or_insert_text<N: Into<Arc<str>>>(&mut self, name: N) -> TextRef { |
| 305 | TextRef::root(name).get_or_create(self) |
| 306 | } |
| 307 | |
| 308 | /// Returns a [MapRef] data structure stored under a given `name`. Maps are used to store key-value |
| 309 | /// pairs associated. These values can be primitive data (similar but not limited to |
| 310 | /// a JavaScript Object Notation) as well as other shared types (Yrs maps, arrays, text |
| 311 | /// structures etc.), enabling to construct a complex recursive tree structures. |
| 312 | /// |
| 313 | /// If no structure under defined `name` existed before, it will be created and returned |
| 314 | /// instead. |
| 315 | /// |
| 316 | /// If a structure under defined `name` already existed, but its type was different it will be |
| 317 | /// reinterpreted as a map (in such case a map component of complex data type will be |
| 318 | /// interpreted as native map). |
| 319 | fn get_or_insert_map<N: Into<Arc<str>>>(&mut self, name: N) -> MapRef { |
| 320 | MapRef::root(name).get_or_create(self) |
| 321 | } |
| 322 | |
| 323 | /// Returns an [ArrayRef] data structure stored under a given `name`. Array structures are used for |
| 324 | /// storing a sequences of elements in ordered manner, positioning given element accordingly |
| 325 | /// to its index. |
| 326 | /// |
| 327 | /// If no structure under defined `name` existed before, it will be created and returned |
| 328 | /// instead. |
| 329 | /// |
| 330 | /// If a structure under defined `name` already existed, but its type was different it will be |
| 331 | /// reinterpreted as an array (in such case a sequence component of complex data type will be |
| 332 | /// interpreted as a list of inserted values). |
| 333 | fn get_or_insert_array<N: Into<Arc<str>>>(&mut self, name: N) -> ArrayRef { |
| 334 | ArrayRef::root(name).get_or_create(self) |
| 335 | } |
| 336 | |
| 337 | /// Returns a [XmlFragmentRef] data structure stored under a given `name`. XML elements represent |
| 338 | /// nodes of XML document. They can contain attributes (key-value pairs, both of string type) |
| 339 | /// as well as other nested XML elements or text values, which are stored in their insertion |
| 340 | /// order. |
| 341 | /// |
| 342 | /// If no structure under defined `name` existed before, it will be created and returned |
| 343 | /// instead. |
| 344 | /// |
| 345 | /// If a structure under defined `name` already existed, but its type was different it will be |
| 346 | /// reinterpreted as a XML element (in such case a map component of complex data type will be |
| 347 | /// interpreted as map of its attributes, while a sequence component - as a list of its child |
| 348 | /// XML nodes). |
| 349 | fn get_or_insert_xml_fragment<N: Into<Arc<str>>>(&mut self, name: N) -> XmlFragmentRef { |
| 350 | XmlFragmentRef::root(name).get_or_create(self) |
| 351 | } |
| 352 | |
| 353 | /// Prunes a pending updates from the current document and returns them. |
| 354 | /// Returns `None` if current document didn't have any pending updates. |
| 355 | fn prune_pending(&mut self) -> Option<Update> { |
| 356 | let mut merge = Vec::with_capacity(2); |
| 357 | let store = self.store_mut(); |
| 358 | if let Some(pending) = store.pending.take() { |
| 359 | merge.push(pending.update); |
| 360 | } |
| 361 | if let Some(pending_ds) = store.pending_ds.take() { |
| 362 | let mut u = Update::new(); |
| 363 | u.delete_set = pending_ds.clone(); |
| 364 | merge.push(u); |
| 365 | } |
| 366 | if merge.is_empty() { |
| 367 | None |
| 368 | } else { |
| 369 | Some(Update::merge_updates(merge)) |
| 370 | } |
| 371 | } |
| 372 | } |
| 373 | |
| 374 | fn merge_pending_v1(update: Vec<u8>, store: &Store) -> Vec<u8> { |
| 375 | let mut merge = VecDeque::new(); |
| 376 | if let Some(pending) = store.pending.as_ref() { |
| 377 | merge.push_back(pending.update.encode_v1()); |
| 378 | } |
| 379 | if let Some(pending_ds) = store.pending_ds.as_ref() { |
| 380 | let mut u = Update::new(); |
| 381 | u.delete_set = pending_ds.clone(); |
| 382 | merge.push_back(u.encode_v1()); |
| 383 | } |
| 384 | if merge.is_empty() { |
| 385 | update |
| 386 | } else { |
| 387 | merge.push_front(update); |
| 388 | merge_updates_v1(merge).unwrap() |
| 389 | } |
| 390 | } |
| 391 | |
| 392 | fn merge_pending_v2(update: Vec<u8>, store: &Store) -> Vec<u8> { |
| 393 | let mut merge = VecDeque::new(); |
| 394 | if let Some(pending) = store.pending.as_ref() { |
| 395 | merge.push_back(pending.update.encode_v2()); |
| 396 | } |
| 397 | if let Some(pending_ds) = store.pending_ds.as_ref() { |
| 398 | let mut u = Update::new(); |
| 399 | u.delete_set = pending_ds.clone(); |
| 400 | merge.push_back(u.encode_v2()); |
| 401 | } |
| 402 | if merge.is_empty() { |
| 403 | update |
| 404 | } else { |
| 405 | merge.push_front(update); |
| 406 | merge_updates_v2(merge).unwrap() |
| 407 | } |
| 408 | } |
| 409 | |
| 410 | /// A very lightweight read-only transaction. These transactions are guaranteed to not modify the |
| 411 | /// contents of an underlying [Doc] and can be used to read it or for serialization purposes. |
| 412 | /// For this reason it's allowed to have a multiple active read-only transactions, but it's |
| 413 | /// not allowed to have any active [read-write transactions](TransactionMut) at the same time. |
| 414 | #[derive(Debug)] |
| 415 | pub struct Transaction<'doc> { |
| 416 | store: RwLockReadGuard<'doc, Store>, |
| 417 | } |
| 418 | |
| 419 | impl<'doc> Transaction<'doc> { |
| 420 | pub(crate) fn new(store: RwLockReadGuard<'doc, Store>) -> Self { |
| 421 | Transaction { store } |
| 422 | } |
| 423 | } |
| 424 | |
| 425 | impl<'doc> ReadTxn for Transaction<'doc> { |
| 426 | #[inline] |
| 427 | fn store(&self) -> &Store { |
| 428 | self.store.deref() |
| 429 | } |
| 430 | } |
| 431 | |
| 432 | /// Read-write transaction. It can be used to modify an underlying state of the corresponding [Doc]. |
| 433 | /// Read-write transactions require an exclusive access to document store - only one such |
| 434 | /// transaction can be present per [Doc] at the same time (read-only [Transaction]s are not allowed |
| 435 | /// to coexists at the same time as well). |
| 436 | /// |
| 437 | /// This transaction type stores the information about all of the changes performed in its scope. |
| 438 | /// These will be used during [TransactionMut::commit] call to optimize metadata of incoming updates, |
| 439 | /// triggering necessary event callbacks etc. For performance reasons it's preferred to batch as |
| 440 | /// many updates as possible using the same transaction. |
| 441 | /// |
| 442 | /// In Yrs transactions are always auto-committing all of their changes when dropped. Rollbacks are |
| 443 | /// not supported (if some operations needs to be undone, this can be achieved using [UndoManager]) |
| 444 | pub struct TransactionMut<'doc> { |
| 445 | pub(crate) store: RwLockWriteGuard<'doc, Store>, |
| 446 | /// State vector of a current transaction at the moment of its creation. |
| 447 | pub(crate) before_state: StateVector, |
| 448 | /// Current state vector of a transaction, which includes all performed updates. |
| 449 | pub(crate) after_state: StateVector, |
| 450 | /// ID's of the blocks to be merged. |
| 451 | pub(crate) merge_blocks: Vec<ID>, |
| 452 | /// Describes the set of deleted items by ids. |
| 453 | pub(crate) delete_set: DeleteSet, |
| 454 | /// We store the reference that last moved an item. This is needed to compute the delta |
| 455 | /// when multiple ContentMove move the same item. |
| 456 | pub(crate) prev_moved: HashMap<ItemPtr, ItemPtr>, |
| 457 | /// All types that were directly modified (property added or child inserted/deleted). |
| 458 | /// New types are not included in this Set. |
| 459 | pub(crate) changed: HashMap<TypePtr, HashSet<Option<Arc<str>>>>, |
| 460 | pub(crate) changed_parent_types: Vec<BranchPtr>, |
| 461 | pub(crate) subdocs: Option<Box<Subdocs>>, |
| 462 | pub(crate) origin: Option<Origin>, |
| 463 | doc: Doc, |
| 464 | committed: bool, |
| 465 | } |
| 466 | |
| 467 | impl<'doc> ReadTxn for TransactionMut<'doc> { |
| 468 | #[inline] |
| 469 | fn store(&self) -> &Store { |
| 470 | self.store.deref() |
| 471 | } |
| 472 | } |
| 473 | |
| 474 | impl<'doc> WriteTxn for TransactionMut<'doc> { |
| 475 | #[inline] |
| 476 | fn store_mut(&mut self) -> &mut Store { |
| 477 | self.store.deref_mut() |
| 478 | } |
| 479 | |
| 480 | fn subdocs_mut(&mut self) -> &mut Subdocs { |
| 481 | self.subdocs.get_or_init() |
| 482 | } |
| 483 | } |
| 484 | |
| 485 | impl<'doc> Drop for TransactionMut<'doc> { |
| 486 | fn drop(&mut self) { |
| 487 | self.commit() |
| 488 | } |
| 489 | } |
| 490 | |
| 491 | impl<'doc> TransactionMut<'doc> { |
| 492 | pub(crate) fn new( |
| 493 | doc: Doc, |
| 494 | store: RwLockWriteGuard<'doc, Store>, |
| 495 | origin: Option<Origin>, |
| 496 | ) -> Self { |
| 497 | let begin_timestamp = store.blocks.get_state_vector(); |
| 498 | TransactionMut { |
| 499 | store, |
| 500 | doc, |
| 501 | origin, |
| 502 | before_state: begin_timestamp, |
| 503 | merge_blocks: Vec::default(), |
| 504 | delete_set: DeleteSet::new(), |
| 505 | after_state: StateVector::default(), |
| 506 | changed: HashMap::default(), |
| 507 | changed_parent_types: Vec::default(), |
| 508 | prev_moved: HashMap::default(), |
| 509 | subdocs: None, |
| 510 | committed: false, |
| 511 | } |
| 512 | } |
| 513 | |
| 514 | pub fn doc(&self) -> &Doc { |
| 515 | &self.doc |
| 516 | } |
| 517 | |
| 518 | pub fn events(&self) -> Option<&StoreEvents> { |
| 519 | self.store.events.as_deref() |
| 520 | } |
| 521 | |
| 522 | pub fn events_mut(&mut self) -> &mut StoreEvents { |
| 523 | self.store.events.get_or_init() |
| 524 | } |
| 525 | |
| 526 | /// Corresponding document's state vector at the moment when current transaction was created. |
| 527 | pub fn before_state(&self) -> &StateVector { |
| 528 | &self.before_state |
| 529 | } |
| 530 | |
| 531 | /// State vector of the transaction after [Transaction::commit] has been called. |
| 532 | pub fn after_state(&self) -> &StateVector { |
| 533 | &self.after_state |
| 534 | } |
| 535 | |
| 536 | /// Data about deletions performed in the scope of current transaction. |
| 537 | pub fn delete_set(&self) -> &DeleteSet { |
| 538 | &self.delete_set |
| 539 | } |
| 540 | |
| 541 | /// Returns origin of the transaction if any was defined. Read-write transactions can get an |
| 542 | /// origin assigned via [Transact::try_transact_mut_with]/[Transact::transact_mut_with] methods. |
| 543 | pub fn origin(&self) -> Option<&Origin> { |
| 544 | self.origin.as_ref() |
| 545 | } |
| 546 | |
| 547 | /// Returns a list of root level types changed in a scope of the current transaction. This |
| 548 | /// list is not filled right away, but as a part of [TransactionMut::commit] process. |
| 549 | pub fn changed_parent_types(&self) -> &[BranchPtr] { |
| 550 | &self.changed_parent_types |
| 551 | } |
| 552 | |
| 553 | #[inline] |
| 554 | pub(crate) fn store(&self) -> &Store { |
| 555 | &self.store |
| 556 | } |
| 557 | |
| 558 | #[inline] |
| 559 | pub(crate) fn store_mut(&mut self) -> &mut Store { |
| 560 | &mut self.store |
| 561 | } |
| 562 | |
| 563 | /// Encodes changes made within the scope of the current transaction using lib0 v1 encoding. |
| 564 | /// |
| 565 | /// Document updates are idempotent and commutative. Caveats: |
| 566 | /// * It doesn't matter in which order document updates are applied. |
| 567 | /// * As long as all clients receive the same document updates, all clients |
| 568 | /// end up with the same content. |
| 569 | /// * Even if an update contains known information, the unknown information |
| 570 | /// is extracted and integrated into the document structure. |
| 571 | pub fn encode_update_v1(&self) -> Vec<u8> { |
| 572 | let mut encoder = EncoderV1::new(); |
| 573 | self.encode_update(&mut encoder); |
| 574 | encoder.to_vec() |
| 575 | } |
| 576 | |
| 577 | /// Encodes changes made within the scope of the current transaction using lib0 v2 encoding. |
| 578 | /// |
| 579 | /// Document updates are idempotent and commutative. Caveats: |
| 580 | /// * It doesn't matter in which order document updates are applied. |
| 581 | /// * As long as all clients receive the same document updates, all clients |
| 582 | /// end up with the same content. |
| 583 | /// * Even if an update contains known information, the unknown information |
| 584 | /// is extracted and integrated into the document structure. |
| 585 | pub fn encode_update_v2(&self) -> Vec<u8> { |
| 586 | let mut encoder = EncoderV2::new(); |
| 587 | self.encode_update(&mut encoder); |
| 588 | encoder.to_vec() |
| 589 | } |
| 590 | |
| 591 | /// Encodes changes made within the scope of the current transaction. |
| 592 | /// |
| 593 | /// Document updates are idempotent and commutative. Caveats: |
| 594 | /// * It doesn't matter in which order document updates are applied. |
| 595 | /// * As long as all clients receive the same document updates, all clients |
| 596 | /// end up with the same content. |
| 597 | /// * Even if an update contains known information, the unknown information |
| 598 | /// is extracted and integrated into the document structure. |
| 599 | pub fn encode_update<E: Encoder>(&self, encoder: &mut E) { |
| 600 | let store = self.store(); |
| 601 | store.write_blocks_from(&self.before_state, encoder); |
| 602 | self.delete_set.encode(encoder); |
| 603 | } |
| 604 | |
| 605 | /// Applies given `id_set` onto current transaction to run multi-range deletion. |
| 606 | /// Returns a remaining of original ID set, that couldn't be applied. |
| 607 | pub(crate) fn apply_delete(&mut self, ds: &DeleteSet) -> Option<DeleteSet> { |
| 608 | let mut unapplied = DeleteSet::new(); |
| 609 | for (client, ranges) in ds.iter() { |
| 610 | if let Some(mut blocks) = self.store.blocks.get_client_mut(client) { |
| 611 | let state = blocks.clock(); |
| 612 | |
| 613 | for range in ranges.iter() { |
| 614 | let clock = range.start; |
| 615 | let clock_end = range.end; |
| 616 | |
| 617 | if clock < state { |
| 618 | if state < clock_end { |
| 619 | unapplied.insert(ID::new(*client, state), clock_end - state); |
| 620 | } |
| 621 | // We can ignore the case of GC and Delete structs, because we are going to skip them |
| 622 | if let Some(mut index) = blocks.find_pivot(clock) { |
| 623 | // We can ignore the case of GC and Delete structs, because we are going to skip them |
| 624 | let ptr = &mut blocks[index]; |
| 625 | if let Some(item) = ptr.as_item() { |
| 626 | // split the first item if necessary |
| 627 | if !item.is_deleted() && item.id.clock < clock { |
| 628 | if let Some(split) = self |
| 629 | .store |
| 630 | .blocks |
| 631 | .split_block_inner(item, clock - item.id.clock) |
| 632 | { |
| 633 | if item.moved.is_some() { |
| 634 | if let Some(&prev_moved) = self.prev_moved.get(&item) { |
| 635 | self.prev_moved.insert(split, prev_moved); |
| 636 | } |
| 637 | } |
| 638 | |
| 639 | index += 1; |
| 640 | self.merge_blocks.push(*split.id()); |
| 641 | } |
| 642 | blocks = self.store.blocks.get_client_mut(client).unwrap(); |
| 643 | } |
| 644 | } |
| 645 | while index < blocks.len() { |
| 646 | let block = &mut blocks[index]; |
| 647 | if let Some(item) = block.as_item() { |
| 648 | if item.id.clock < clock_end { |
| 649 | if !item.is_deleted() { |
| 650 | if item.id.clock + item.len() > clock_end { |
| 651 | if let Some(split) = |
| 652 | self.store.blocks.split_block_inner( |
| 653 | item, |
| 654 | clock_end - item.id.clock, |
| 655 | ) |
| 656 | { |
| 657 | if item.moved.is_some() { |
| 658 | if let Some(&prev_moved) = |
| 659 | self.prev_moved.get(&item) |
| 660 | { |
| 661 | self.prev_moved |
| 662 | .insert(split, prev_moved); |
| 663 | } |
| 664 | } |
| 665 | if item.info.is_linked() { |
| 666 | if let Some(links) = self |
| 667 | .store |
| 668 | .linked_by |
| 669 | .get(&item) |
| 670 | .cloned() |
| 671 | { |
| 672 | self.store |
| 673 | .linked_by |
| 674 | .insert(split, links); |
| 675 | } |
| 676 | } |
| 677 | |
| 678 | self.merge_blocks.push(*split.id()); |
| 679 | index += 1; |
| 680 | } |
| 681 | } |
| 682 | self.delete(item); |
| 683 | blocks = self |
| 684 | .store |
| 685 | .blocks |
| 686 | .get_client_mut(client) |
| 687 | .unwrap(); |
| 688 | // just to make the borrow checker happy |
| 689 | } |
| 690 | } else { |
| 691 | break; |
| 692 | } |
| 693 | } |
| 694 | index += 1; |
| 695 | } |
| 696 | } |
| 697 | } else { |
| 698 | unapplied.insert(ID::new(*client, clock), clock_end - clock); |
| 699 | } |
| 700 | } |
| 701 | } else { |
| 702 | for range in ranges.iter() { |
| 703 | unapplied.insert(ID::new(*client, range.start), range.end - range.start); |
| 704 | } |
| 705 | } |
| 706 | } |
| 707 | |
| 708 | if unapplied.is_empty() { |
| 709 | None |
| 710 | } else { |
| 711 | Some(unapplied) |
| 712 | } |
| 713 | } |
| 714 | |
| 715 | /// Delete item under given pointer. |
| 716 | /// Returns true if block was successfully deleted, false if it was already deleted in the past. |
| 717 | pub(crate) fn delete(&mut self, mut item: ItemPtr) -> bool { |
| 718 | let mut recurse = Vec::new(); |
| 719 | let mut result = false; |
| 720 | |
| 721 | let ptr = item.clone(); |
| 722 | let store = self.store.deref(); |
| 723 | if !item.is_deleted() { |
| 724 | if item.parent_sub.is_none() && item.is_countable() { |
| 725 | if let TypePtr::Branch(mut parent) = item.parent { |
| 726 | parent.block_len -= item.len(); |
| 727 | parent.content_len -= item.content_len(store.offset_kind); |
| 728 | } |
| 729 | } |
| 730 | |
| 731 | item.mark_as_deleted(); |
| 732 | self.delete_set.insert(item.id.clone(), item.len()); |
| 733 | if let Some(parent) = item.parent.as_branch() { |
| 734 | self.add_changed_type(*parent, item.parent_sub.clone()); |
| 735 | } else { |
| 736 | // parent has been GC'ed |
| 737 | } |
| 738 | |
| 739 | match &mut item.content { |
| 740 | ItemContent::Doc(_, doc) => { |
| 741 | let subdocs = self.subdocs.get_or_init(); |
| 742 | let addr = doc.addr(); |
| 743 | if subdocs.added.remove(&addr).is_none() { |
| 744 | subdocs.removed.insert(addr, doc.clone()); |
| 745 | } |
| 746 | } |
| 747 | ItemContent::Type(inner) => { |
| 748 | let branch_ptr = BranchPtr::from(inner); |
| 749 | #[cfg(feature = "weak")] |
| 750 | if let crate::types::TypeRef::WeakLink(source) = &branch_ptr.type_ref { |
| 751 | source.unlink_all(self, branch_ptr); |
| 752 | } |
| 753 | let mut ptr = branch_ptr.start; |
| 754 | self.changed.remove(&TypePtr::Branch(branch_ptr)); |
| 755 | |
| 756 | while let Some(item) = ptr.as_deref() { |
| 757 | if !item.is_deleted() { |
| 758 | recurse.push(ptr.unwrap()); |
| 759 | } |
| 760 | |
| 761 | ptr = item.right.clone(); |
| 762 | } |
| 763 | |
| 764 | for ptr in branch_ptr.map.values() { |
| 765 | recurse.push(ptr.clone()); |
| 766 | } |
| 767 | } |
| 768 | ItemContent::Move(m) => m.delete(self, ptr), |
| 769 | _ => { /* nothing to do for other content types */ } |
| 770 | } |
| 771 | if item.info.is_linked() { |
| 772 | // notify links that current element has been removed |
| 773 | if let Some(linked_by) = self.store.linked_by.remove(&item) { |
| 774 | for link in linked_by { |
| 775 | self.add_changed_type(link, item.parent_sub.clone()); |
| 776 | } |
| 777 | } |
| 778 | } |
| 779 | result = true; |
| 780 | } |
| 781 | |
| 782 | for &ptr in recurse.iter() { |
| 783 | let id = *ptr.id(); |
| 784 | if !self.delete(ptr) { |
| 785 | // Whis will be gc'd later and we want to merge it if possible |
| 786 | // We try to merge all deleted items after each transaction, |
| 787 | // but we have no knowledge about that this needs to be merged |
| 788 | // since it is not in transaction.ds. Hence we add it to transaction._mergeStructs |
| 789 | self.merge_blocks.push(id); |
| 790 | } |
| 791 | } |
| 792 | |
| 793 | result |
| 794 | } |
| 795 | |
| 796 | /// Applies a deserialized [Update] contents into a document owning current transaction. Update |
| 797 | /// payload can be generated by methods such as [TransactionMut::encode_diff] or passed to |
| 798 | /// [Doc::observe_update_v1]/[Doc::observe_update_v2] callbacks. Updates are allowed to contain |
| 799 | /// duplicate blocks (already presen in current document store) - these will be ignored. |
| 800 | /// |
| 801 | /// # Pending updates |
| 802 | /// |
| 803 | /// Remote update integration requires that all to-be-integrated blocks must have their direct |
| 804 | /// predecessors already in place. Out of order updates from the same peer will be stashed |
| 805 | /// internally and their integration will be postponed until missing blocks arrive first. |
| 806 | pub fn apply_update(&mut self, update: Update) -> Result<(), UpdateError> { |
| 807 | let (remaining, remaining_ds) = update.integrate(self)?; |
| 808 | let mut retry = false; |
| 809 | { |
| 810 | let store = self.store_mut(); |
| 811 | store.pending = if let Some(mut pending) = store.pending.take() { |
| 812 | // check if we can apply something |
| 813 | for (client, &clock) in pending.missing.iter() { |
| 814 | if clock < store.blocks.get_clock(client) { |
| 815 | retry = true; |
| 816 | break; |
| 817 | } |
| 818 | } |
| 819 | |
| 820 | if let Some(remaining) = remaining { |
| 821 | // merge restStructs into store.pending |
| 822 | for (&client, &clock) in remaining.missing.iter() { |
| 823 | pending.missing.set_min(client, clock); |
| 824 | } |
| 825 | pending.update = Update::merge_updates(vec![pending.update, remaining.update]); |
| 826 | } |
| 827 | Some(pending) |
| 828 | } else { |
| 829 | remaining |
| 830 | }; |
| 831 | } |
| 832 | if let Some(pending) = self.store_mut().pending_ds.take() { |
| 833 | let ds2 = self.apply_delete(&pending); |
| 834 | let ds = match (remaining_ds, ds2) { |
| 835 | (Some(mut a), Some(b)) => { |
| 836 | a.delete_set.merge(b); |
| 837 | Some(a.delete_set) |
| 838 | } |
| 839 | (Some(x), _) => Some(x.delete_set), |
| 840 | (_, Some(x)) => Some(x), |
| 841 | _ => None, |
| 842 | }; |
| 843 | self.store_mut().pending_ds = ds; |
| 844 | } else { |
| 845 | self.store_mut().pending_ds = remaining_ds.map(|update| update.delete_set); |
| 846 | } |
| 847 | |
| 848 | if retry { |
| 849 | let store = self.store_mut(); |
| 850 | if let Some(pending) = store.pending.take() { |
| 851 | let ds = store.pending_ds.take().unwrap_or_default(); |
| 852 | let mut ds_update = Update::new(); |
| 853 | ds_update.delete_set = ds; |
| 854 | self.apply_update(pending.update)?; |
| 855 | self.apply_update(ds_update)?; |
| 856 | } |
| 857 | } |
| 858 | |
| 859 | Ok(()) |
| 860 | } |
| 861 | |
| 862 | pub(crate) fn create_item<T: Prelim>( |
| 863 | &mut self, |
| 864 | pos: &ItemPosition, |
| 865 | value: T, |
| 866 | parent_sub: Option<Arc<str>>, |
| 867 | ) -> Option<ItemPtr> { |
| 868 | let (left, right, origin, id) = { |
| 869 | let store = self.store_mut(); |
| 870 | let left = pos.left; |
| 871 | let right = pos.right; |
| 872 | let origin = if let Some(item) = pos.left.as_deref() { |
| 873 | Some(item.last_id()) |
| 874 | } else { |
| 875 | None |
| 876 | }; |
| 877 | let client_id = store.client_id; |
| 878 | let id = ID::new(client_id, store.get_local_state()); |
| 879 | |
| 880 | (left, right, origin, id) |
| 881 | }; |
| 882 | let (mut content, remainder) = value.into_content(self); |
| 883 | let inner_ref = if let ItemContent::Type(inner_ref) = &mut content { |
| 884 | Some(BranchPtr::from(inner_ref)) |
| 885 | } else { |
| 886 | None |
| 887 | }; |
| 888 | let mut block = Item::new( |
| 889 | id, |
| 890 | left, |
| 891 | origin, |
| 892 | right, |
| 893 | right.map(|r| r.id().clone()), |
| 894 | pos.parent.clone(), |
| 895 | parent_sub, |
| 896 | content, |
| 897 | )?; |
| 898 | let mut block_ptr = ItemPtr::from(&mut block); |
| 899 | |
| 900 | block_ptr.integrate(self, 0); |
| 901 | |
| 902 | self.store_mut().blocks.push_block(block); |
| 903 | |
| 904 | if let Some(remainder) = remainder { |
| 905 | remainder.integrate(self, inner_ref.unwrap().into()) |
| 906 | } |
| 907 | |
| 908 | Some(block_ptr) |
| 909 | } |
| 910 | |
| 911 | fn call_type_observers( |
| 912 | changed_parent_types: &mut Vec<BranchPtr>, |
| 913 | all_links: &HashMap<ItemPtr, HashSet<BranchPtr>>, |
| 914 | branch: BranchPtr, |
| 915 | changed_parents: &mut HashMap<BranchPtr, Vec<usize>>, |
| 916 | event_cache: &Vec<Event>, |
| 917 | visited: &mut HashSet<BranchPtr>, |
| 918 | ) { |
| 919 | let mut current = branch; |
| 920 | loop { |
| 921 | changed_parent_types.push(current); |
| 922 | if current.deep_observers.has_subscribers() { |
| 923 | let entries = changed_parents.entry(current).or_default(); |
| 924 | entries.push(event_cache.len() - 1); |
| 925 | } |
| 926 | |
| 927 | if let Some(item) = current.item { |
| 928 | if item.info.is_linked() { |
| 929 | if let Some(linked_by) = all_links.get(&item) { |
| 930 | for &link in linked_by.iter() { |
| 931 | if visited.insert(link) { |
| 932 | Self::call_type_observers( |
| 933 | changed_parent_types, |
| 934 | all_links, |
| 935 | link, |
| 936 | changed_parents, |
| 937 | event_cache, |
| 938 | visited, |
| 939 | ) |
| 940 | } |
| 941 | } |
| 942 | } |
| 943 | } |
| 944 | if let TypePtr::Branch(parent) = item.parent { |
| 945 | current = parent; |
| 946 | continue; |
| 947 | } |
| 948 | } |
| 949 | |
| 950 | break; |
| 951 | } |
| 952 | } |
| 953 | |
| 954 | /// Commits current transaction. This step involves cleaning up and optimizing changes performed |
| 955 | /// during lifetime of a transaction. Such changes include squashing delete sets data, |
| 956 | /// squashing blocks that have been appended one after another to preserve memory and triggering |
| 957 | /// events. |
| 958 | /// |
| 959 | /// This step is performed automatically when a transaction is about to be dropped (its life |
| 960 | /// scope comes to an end). |
| 961 | pub fn commit(&mut self) { |
| 962 | if self.committed { |
| 963 | return; |
| 964 | } |
| 965 | self.committed = true; |
| 966 | |
| 967 | // 1. sort and merge delete set |
| 968 | self.delete_set.squash(); |
| 969 | self.after_state = self.store.blocks.get_state_vector(); |
| 970 | // 2. emit 'beforeObserverCalls' |
| 971 | // 3. for each change observed by the transaction call 'afterTransaction' |
| 972 | if !self.changed.is_empty() { |
| 973 | let mut changed_parents: HashMap<BranchPtr, Vec<usize>> = HashMap::new(); |
| 974 | let mut event_cache = Vec::new(); |
| 975 | |
| 976 | for (ptr, subs) in self.changed.iter() { |
| 977 | if let TypePtr::Branch(branch) = ptr { |
| 978 | if let Some(e) = branch.trigger(self, subs.clone()) { |
| 979 | event_cache.push(e); |
| 980 | Self::call_type_observers( |
| 981 | &mut self.changed_parent_types, |
| 982 | &self.store.linked_by, |
| 983 | *branch, |
| 984 | &mut changed_parents, |
| 985 | &event_cache, |
| 986 | &mut HashSet::default(), |
| 987 | ); |
| 988 | } |
| 989 | } |
| 990 | } |
| 991 | |
| 992 | // deep observe events |
| 993 | for (&branch, events) in changed_parents.iter() { |
| 994 | // sort events by path length so that top-level events are fired first. |
| 995 | let mut unsorted: Vec<&Event> = Vec::with_capacity(events.len()); |
| 996 | |
| 997 | for &i in events.iter() { |
| 998 | let e = &mut event_cache[i]; |
| 999 | e.set_current_target(branch); |
| 1000 | } |
| 1001 | |
| 1002 | for &i in events.iter() { |
| 1003 | unsorted.push(&event_cache[i]); |
| 1004 | } |
| 1005 | |
| 1006 | // We don't need to check for events.length |
| 1007 | // because we know it has at least one element |
| 1008 | let events = Events::new(&mut unsorted); |
| 1009 | branch.trigger_deep(self, &events); |
| 1010 | } |
| 1011 | } |
| 1012 | |
| 1013 | if let Some(events) = self.store.events.take() { |
| 1014 | events.emit_after_transaction(self); |
| 1015 | self.store.events = Some(events); |
| 1016 | } |
| 1017 | |
| 1018 | // 4. try GC delete set |
| 1019 | if !self.store.skip_gc { |
| 1020 | GCCollector::collect(self); |
| 1021 | } |
| 1022 | |
| 1023 | // 5. try merge delete set |
| 1024 | self.delete_set.try_squash_with(&mut self.store); |
| 1025 | |
| 1026 | // 6. get transaction after state and try to merge to left |
| 1027 | for (client, &clock) in self.after_state.iter() { |
| 1028 | let before_clock = self.before_state.get(client); |
| 1029 | if before_clock != clock { |
| 1030 | let blocks = self.store.blocks.get_client_mut(client).unwrap(); |
| 1031 | let first_change = blocks.find_pivot(before_clock).unwrap().max(1); |
| 1032 | let mut i = blocks.len() - 1; |
| 1033 | while i >= first_change { |
| 1034 | blocks.squash_left(i); |
| 1035 | i -= 1; |
| 1036 | } |
| 1037 | } |
| 1038 | } |
| 1039 | |
| 1040 | // 7. get merge_structs and try to merge to left |
| 1041 | for id in self.merge_blocks.iter() { |
| 1042 | if let Some(blocks) = self.store.blocks.get_client_mut(&id.client) { |
| 1043 | if let Some(replaced_pos) = blocks.find_pivot(id.clock) { |
| 1044 | if replaced_pos + 1 < blocks.len() { |
| 1045 | blocks.squash_left(replaced_pos + 1); |
| 1046 | } else if replaced_pos > 0 { |
| 1047 | blocks.squash_left(replaced_pos); |
| 1048 | } |
| 1049 | } |
| 1050 | } |
| 1051 | } |
| 1052 | |
| 1053 | if let Some(events) = self.store.events.as_ref() { |
| 1054 | // 8. emit 'afterTransactionCleanup' |
| 1055 | events.emit_transaction_cleanup(self); |
| 1056 | // 9. emit 'update' |
| 1057 | events.emit_update_v1(self); |
| 1058 | // 10. emit 'updateV2' |
| 1059 | events.emit_update_v2(self); |
| 1060 | } |
| 1061 | |
| 1062 | // 11. add and remove subdocs |
| 1063 | let store = self.store.deref_mut(); |
| 1064 | if let Some(mut subdocs) = self.subdocs.take() { |
| 1065 | let client_id = store.client_id; |
| 1066 | for (guid, subdoc) in subdocs.added.iter_mut() { |
| 1067 | let mut txn = subdoc.transact_mut(); |
| 1068 | txn.store.client_id = client_id; |
| 1069 | txn.doc |
| 1070 | .store() |
| 1071 | .set_subdoc_data(client_id, self.doc.collection_id()); |
| 1072 | store.subdocs.insert(guid.clone(), subdoc.clone()); |
| 1073 | } |
| 1074 | for guid in subdocs.removed.keys() { |
| 1075 | store.subdocs.remove(guid); |
| 1076 | } |
| 1077 | |
| 1078 | let store = self.store.deref(); |
| 1079 | let mut removed = if let Some(events) = store.events.as_ref() { |
| 1080 | if events.subdocs_events.has_subscribers() { |
| 1081 | let e = SubdocsEvent::new(subdocs); |
| 1082 | events.subdocs_events.trigger(|cb| cb(self, &e)); |
| 1083 | e.removed |
| 1084 | } else { |
| 1085 | subdocs.removed |
| 1086 | } |
| 1087 | } else { |
| 1088 | subdocs.removed |
| 1089 | }; |
| 1090 | |
| 1091 | for (_, subdoc) in removed.iter_mut() { |
| 1092 | subdoc.destroy(self); |
| 1093 | } |
| 1094 | } |
| 1095 | } |
| 1096 | |
| 1097 | /// Perform garbage collection of deleted blocks, even if a document was created with `skip_gc` |
| 1098 | /// option. This operation will scan over ALL deleted elements, NOT ONLY the ones that have been |
| 1099 | /// changed as part of this transaction scope. |
| 1100 | pub fn force_gc(&mut self) { |
| 1101 | GCCollector::collect_all(self); |
| 1102 | } |
| 1103 | |
| 1104 | pub(crate) fn add_changed_type(&mut self, parent: BranchPtr, parent_sub: Option<Arc<str>>) { |
| 1105 | let trigger = if let Some(ptr) = parent.item { |
| 1106 | (ptr.id().clock < self.before_state.get(&ptr.id().client)) && !ptr.is_deleted() |
| 1107 | } else { |
| 1108 | true |
| 1109 | }; |
| 1110 | if trigger { |
| 1111 | let e = self.changed.entry(parent.into()).or_default(); |
| 1112 | e.insert(parent_sub.clone()); |
| 1113 | } |
| 1114 | } |
| 1115 | |
| 1116 | /// Checks if item with a given `id` has been added to a block store within this transaction. |
| 1117 | pub(crate) fn has_added(&self, id: &ID) -> bool { |
| 1118 | id.clock >= self.before_state.get(&id.client) |
| 1119 | } |
| 1120 | |
| 1121 | /// Checks if item with a given `id` has been deleted within this transaction. |
| 1122 | pub(crate) fn has_deleted(&self, id: &ID) -> bool { |
| 1123 | self.delete_set.is_deleted(id) |
| 1124 | } |
| 1125 | |
| 1126 | pub(crate) fn split_by_snapshot(&mut self, snapshot: &Snapshot) { |
| 1127 | let mut merge_blocks: Vec<ID> = Vec::new(); |
| 1128 | let blocks = &mut self.store.blocks; |
| 1129 | for (&client, &clock) in snapshot.state_map.iter() { |
| 1130 | if let Some(ptr) = blocks.get_item(&ID::new(client, clock)) { |
| 1131 | let ptr_clock = ptr.id.clock; |
| 1132 | if ptr_clock < clock { |
| 1133 | if let Some(right) = blocks.split_block_inner(ptr, clock - ptr_clock) { |
| 1134 | if right.moved.is_some() { |
| 1135 | if let Some(&prev_moved) = self.prev_moved.get(&ptr) { |
| 1136 | self.prev_moved.insert(right, prev_moved); |
| 1137 | } |
| 1138 | } |
| 1139 | |
| 1140 | merge_blocks.push(*right.id()); |
| 1141 | } |
| 1142 | } |
| 1143 | } |
| 1144 | } |
| 1145 | |
| 1146 | self.merge_blocks.append(&mut merge_blocks); |
| 1147 | let mut deleted = snapshot.delete_set.deleted_blocks(); |
| 1148 | while let Some(slice) = deleted.next(self) { |
| 1149 | if let BlockSlice::Item(slice) = slice { |
| 1150 | //TODO: we technically don't need to physically split underlying item in two |
| 1151 | // if we were to use block slices all the way down. |
| 1152 | |
| 1153 | // split the blocks by delete set |
| 1154 | let ptr = self.store.materialize(slice); |
| 1155 | self.merge_blocks.push(ptr.id); |
| 1156 | } |
| 1157 | } |
| 1158 | } |
| 1159 | |
| 1160 | #[cfg(feature = "weak")] |
| 1161 | pub(crate) fn unlink(&mut self, mut source: ItemPtr, link: BranchPtr) { |
| 1162 | let all_links = &mut self.store.linked_by; |
| 1163 | let prune = if let Some(linked_by) = all_links.get_mut(&source) { |
| 1164 | linked_by.remove(&link) && linked_by.is_empty() |
| 1165 | } else { |
| 1166 | false |
| 1167 | }; |
| 1168 | if prune { |
| 1169 | all_links.remove(&source); |
| 1170 | source.info.clear_linked(); |
| 1171 | if source.is_countable() { |
| 1172 | // since linked property is blocking items from merging, |
| 1173 | // it may turn out that source item can be merged now |
| 1174 | self.merge_blocks.push(source.id); |
| 1175 | } |
| 1176 | } |
| 1177 | } |
| 1178 | } |
| 1179 | |
| 1180 | /// Iterator struct used to traverse over all of the root level types defined in a corresponding [Doc]. |
| 1181 | pub struct RootRefs<'doc>(std::collections::hash_map::Iter<'doc, Arc<str>, Box<Branch>>); |
| 1182 | |
| 1183 | impl<'doc> Iterator for RootRefs<'doc> { |
| 1184 | type Item = (&'doc str, Out); |
| 1185 | |
| 1186 | fn next(&mut self) -> Option<Self::Item> { |
| 1187 | let (key, branch) = self.0.next()?; |
| 1188 | let key = key.as_ref(); |
| 1189 | let ptr = BranchPtr::from(branch); |
| 1190 | Some((key, ptr.into())) |
| 1191 | } |
| 1192 | } |
| 1193 | |
| 1194 | #[derive(Default)] |
| 1195 | pub struct Subdocs { |
| 1196 | pub(crate) added: HashMap<DocAddr, Doc>, |
| 1197 | pub(crate) removed: HashMap<DocAddr, Doc>, |
| 1198 | pub(crate) loaded: HashMap<DocAddr, Doc>, |
| 1199 | } |
| 1200 | |
| 1201 | /// A binary marker that can be assigned to a read-write transaction upon creation via |
| 1202 | /// [Transact::try_transact_mut_with]/[Transact::transact_mut_with]. It can be used to classify |
| 1203 | /// transaction updates within a specific context, which exists for the duration of a transaction |
| 1204 | /// (it's **not persisted** in the document store itself), i.e. *you can use unique document client |
| 1205 | /// identifiers to differentiate updates incoming from remote nodes from those performed locally*. |
| 1206 | #[repr(transparent)] |
| 1207 | #[derive(Clone, Ord, PartialOrd, Eq, PartialEq, Hash)] |
| 1208 | pub struct Origin(SmallVec<[u8; std::mem::size_of::<usize>()]>); |
| 1209 | |
| 1210 | impl AsRef<[u8]> for Origin { |
| 1211 | fn as_ref(&self) -> &[u8] { |
| 1212 | self.0.as_ref() |
| 1213 | } |
| 1214 | } |
| 1215 | |
| 1216 | impl<'a, T> From<Pin<&'a T>> for Origin { |
| 1217 | fn from(p: Pin<&T>) -> Self { |
| 1218 | let ptr = Pin::get_ref(p) as *const T as usize; |
| 1219 | Origin(SmallVec::from_const(ptr.to_be_bytes())) |
| 1220 | } |
| 1221 | } |
| 1222 | |
| 1223 | impl<'a> From<&'a [u8]> for Origin { |
| 1224 | fn from(slice: &'a [u8]) -> Self { |
| 1225 | Origin(SmallVec::from_slice(slice)) |
| 1226 | } |
| 1227 | } |
| 1228 | |
| 1229 | impl<'a> From<&'a str> for Origin { |
| 1230 | fn from(v: &'a str) -> Self { |
| 1231 | Origin(SmallVec::from_slice(v.as_ref())) |
| 1232 | } |
| 1233 | } |
| 1234 | |
| 1235 | impl From<String> for Origin { |
| 1236 | fn from(v: String) -> Self { |
| 1237 | Origin(SmallVec::from(Vec::from(v))) |
| 1238 | } |
| 1239 | } |
| 1240 | |
| 1241 | impl std::fmt::Debug for Origin { |
| 1242 | fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { |
| 1243 | std::fmt::Display::fmt(self, f) |
| 1244 | } |
| 1245 | } |
| 1246 | |
| 1247 | impl std::fmt::Display for Origin { |
| 1248 | fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { |
| 1249 | write!(f, "Origin(")?; |
| 1250 | for b in self.0.iter() { |
| 1251 | write!(f, "{:02x?}", b)?; |
| 1252 | } |
| 1253 | write!(f, ")") |
| 1254 | } |
| 1255 | } |
| 1256 | |
| 1257 | macro_rules! impl_origin { |
| 1258 | ($t:ty) => { |
| 1259 | impl From<$t> for Origin { |
| 1260 | fn from(v: $t) -> Origin { |
| 1261 | Origin(SmallVec::from_slice(&v.to_be_bytes())) |
| 1262 | } |
| 1263 | } |
| 1264 | }; |
| 1265 | } |
| 1266 | |
| 1267 | impl_origin!(u8); |
| 1268 | impl_origin!(u16); |
| 1269 | impl_origin!(u32); |
| 1270 | impl_origin!(u64); |
| 1271 | impl_origin!(u128); |
| 1272 | impl_origin!(usize); |
| 1273 | impl_origin!(i8); |
| 1274 | impl_origin!(i16); |
| 1275 | impl_origin!(i32); |
| 1276 | impl_origin!(i64); |
| 1277 | impl_origin!(i128); |
| 1278 | impl_origin!(isize); |
| 1279 | |
amadan