Jayden/AFFiNE-Mirror
0
0
Fork 0
mirror of https://github.com/toeverything/AFFiNE.git synced 2026-08-24 05:07:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(native): sync yocto codes (#14243)

Browse Source 
#### PR Dependency Tree


* **PR #14243** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Batch management API for coordinated document mutations and change
tracking.
* New document accessors (IDs, state snapshots, change/delete set
queries) and subscriber count.

* **Chores**
  * Upgraded Rust edition across packages to 2024.
  * Repository-wide formatting, stylistic cleanups and test adjustments.

* **Breaking Changes**
* Removed the Node native bindings package and its JS/TS declarations
and tests (no longer published/available).

<sub>✏️ Tip: You can customize this high-level summary in your review
settings.</sub>
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
DarkSky
2026-01-11 06:08:33 +08:00
committed by GitHub
parent d515d295ce
commit ca2462f987
143 changed files with 1396 additions and 4841 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/common/y-octo/core/src/codec/buffer.rs
+5 -14
View File
@@ -19,8 +19,8 @@ pub fn write_var_buffer<W: Write>(buffer: &mut W, data: &[u8]) -> Result<(), Err
#[cfg(test)]
mod tests {
use nom::{
error::{Error, ErrorKind},
AsBytes, Err,
error::{Error, ErrorKind},
};
use super::*;
@@ -36,26 +36,17 @@ mod tests {
// Test case 2: truncated input, missing buffer
let input = [0x05, 0x01, 0x02, 0x03];
let result = read_var_buffer(&input);
assert_eq!(
result,
Err(Err::Error(Error::new(&input[1..], ErrorKind::Eof)))
);
assert_eq!(result, Err(Err::Error(Error::new(&input[1..], ErrorKind::Eof))));
// Test case 3: invalid input
let input = [0xFF, 0x01, 0x02, 0x03];
let result = read_var_buffer(&input);
assert_eq!(
result,
Err(Err::Error(Error::new(&input[2..], ErrorKind::Eof)))
);
assert_eq!(result, Err(Err::Error(Error::new(&input[2..], ErrorKind::Eof))));
// Test case 4: invalid var int encoding
let input = [0xFF, 0x80, 0x80, 0x80, 0x80, 0x80, 0x01];
let result = read_var_buffer(&input);
assert_eq!(
result,
Err(Err::Error(Error::new(&input[7..], ErrorKind::Eof)))
);
assert_eq!(result, Err(Err::Error(Error::new(&input[7..], ErrorKind::Eof))));
}
#[test]
@@ -66,7 +57,7 @@ mod tests {
#[cfg(not(miri))]
{
use rand::{rng, Rng};
use rand::{Rng, rng};
let mut rng = rng();
for _ in 0..100 {
test_var_buf_enc_dec(&{
packages/common/y-octo/core/src/codec/string.rs
+7 -22
View File
@@ -1,6 +1,6 @@
use std::io::{Error, Write};
use nom::{combinator::map_res, Parser};
use nom::{Parser, combinator::map_res};
use super::*;
@@ -17,8 +17,8 @@ pub fn write_var_string<W: Write, S: AsRef<str>>(buffer: &mut W, input: S) -> Re
#[cfg(test)]
mod tests {
use nom::{
error::{Error, ErrorKind},
AsBytes, Err,
error::{Error, ErrorKind},
};
use super::*;
@@ -34,42 +34,27 @@ mod tests {
// Test case 2: missing string length
let input = [0x68, 0x65, 0x6C, 0x6C, 0x6F];
let result = read_var_string(&input);
assert_eq!(
result,
Err(Err::Error(Error::new(&input[1..], ErrorKind::Eof)))
);
assert_eq!(result, Err(Err::Error(Error::new(&input[1..], ErrorKind::Eof))));
// Test case 3: truncated input
let input = [0x05, 0x68, 0x65, 0x6C, 0x6C];
let result = read_var_string(&input);
assert_eq!(
result,
Err(Err::Error(Error::new(&input[1..], ErrorKind::Eof)))
);
assert_eq!(result, Err(Err::Error(Error::new(&input[1..], ErrorKind::Eof))));
// Test case 4: invalid input
let input = [0xFF, 0x01, 0x02, 0x03, 0x04];
let result = read_var_string(&input);
assert_eq!(
result,
Err(Err::Error(Error::new(&input[2..], ErrorKind::Eof)))
);
assert_eq!(result, Err(Err::Error(Error::new(&input[2..], ErrorKind::Eof))));
// Test case 5: invalid var int encoding
let input = [0xFF, 0x80, 0x80, 0x80, 0x80, 0x80, 0x01];
let result = read_var_string(&input);
assert_eq!(
result,
Err(Err::Error(Error::new(&input[7..], ErrorKind::Eof)))
);
assert_eq!(result, Err(Err::Error(Error::new(&input[7..], ErrorKind::Eof))));
// Test case 6: invalid input, invalid UTF-8 encoding
let input = [0x05, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF];
let result = read_var_string(&input);
assert_eq!(
result,
Err(Err::Error(Error::new(&input[..], ErrorKind::MapRes)))
);
assert_eq!(result, Err(Err::Error(Error::new(&input[..], ErrorKind::MapRes))));
}
#[test]
packages/common/y-octo/core/src/doc/awareness.rs
+10 -23
View File
@@ -20,6 +20,10 @@ impl Awareness {
}
}
pub fn local_id(&self) -> u64 {
self.local_id
}
pub fn on_update(&mut self, f: impl Fn(&Awareness, AwarenessEvent) + Send + Sync + 'static) {
self.callback = Some(Arc::new(f));
}
@@ -29,10 +33,7 @@ impl Awareness {
}
pub fn get_local_state(&self) -> Option<String> {
self
.awareness
.get(&self.local_id)
.map(|state| state.content.clone())
self.awareness.get(&self.local_id).map(|state| state.content.clone())
}
fn mut_local_state(&mut self) -> &mut AwarenessState {
@@ -42,20 +43,14 @@ impl Awareness {
pub fn set_local_state(&mut self, content: String) {
self.mut_local_state().set_content(content);
if let Some(cb) = self.callback.as_ref() {
cb(
self,
AwarenessEventBuilder::new().update(self.local_id).build(),
);
cb(self, AwarenessEventBuilder::new().update(self.local_id).build());
}
}
pub fn clear_local_state(&mut self) {
self.mut_local_state().delete();
if let Some(cb) = self.callback.as_ref() {
cb(
self,
AwarenessEventBuilder::new().remove(self.local_id).build(),
);
cb(self, AwarenessEventBuilder::new().remove(self.local_id).build());
}
}
@@ -107,9 +102,7 @@ impl AwarenessEvent {
pub fn get_updated(&self, states: &AwarenessStates) -> AwarenessStates {
states
.iter()
.filter(|(id, _)| {
self.added.contains(id) || self.updated.contains(id) || self.removed.contains(id)
})
.filter(|(id, _)| self.added.contains(id) || self.updated.contains(id) || self.removed.contains(id))
.map(|(id, state)| (*id, state.clone()))
.collect()
}
@@ -187,14 +180,8 @@ mod tests {
assert!(awareness.get_states().contains_key(&1));
// local state will not apply
assert_eq!(
awareness.get_states().get(&0).unwrap().content,
"null".to_string()
);
assert_eq!(
awareness.get_states().get(&1).unwrap().content,
"test1".to_string()
);
assert_eq!(awareness.get_states().get(&0).unwrap().content, "null".to_string());
assert_eq!(awareness.get_states().get(&1).unwrap().content, "test1".to_string());
}
{
packages/common/y-octo/core/src/doc/batch.rs
+119
View File
@@ -0,0 +1,119 @@
use super::*;
#[derive(Debug, PartialEq)]
pub struct Batch {
doc: Doc,
before_state: StateVector,
after_state: StateVector,
changed: HashMap<YTypeRef, Vec<SmolStr>>,
}
impl Batch {
pub fn new(doc: Doc) -> Self {
let current_state = doc.get_state_vector();
Batch {
doc,
before_state: current_state.clone(),
after_state: current_state,
changed: HashMap::new(),
}
}
pub fn with_batch<T, F>(&mut self, f: F) -> T
where
F: FnOnce(Doc) -> T,
{
let ret = f(self.doc.clone());
for (k, v) in self.doc.get_changed() {
self.changed.entry(k).or_default().extend(v.iter().cloned());
}
ret
}
}
pub fn batch_commit<T, F>(mut doc: Doc, f: F) -> Option<T>
where
F: FnOnce(Doc) -> T,
{
// Initialize batch cleanups list
let mut batch_cleanups = vec![];
// Initial call and result initialization
let mut initial_call = false;
{
if doc.batch.is_none() {
initial_call = true;
// Start a new batch
let batch = Batch::new(doc.clone());
doc.batch = Somr::new(batch);
batch_cleanups.push(doc.batch.clone());
}
}
let batch = doc.batch.get_mut()?;
let result = Some(batch.with_batch(f));
if initial_call
&& let Some(current_batch) = doc.batch.get()
&& Some(current_batch) == batch_cleanups[0].get()
{
// Process observer calls and perform cleanup if this is the initial call
cleanup_batches(&mut batch_cleanups);
doc.batch.swap_take();
}
result
}
fn cleanup_batches(batch_cleanups: &mut Vec<Somr<Batch>>) {
for batch in batch_cleanups.drain(..) {
if let Some(batch) = batch.get() {
println!("changed: {:?}", batch.changed);
} else {
panic!("Batch not initialized");
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn should_get_changed_items() {
loom_model!({
let doc = DocOptions::new().with_client_id(1).build();
batch_commit(doc.clone(), |d| {
let mut arr = d.get_or_create_array("arr").unwrap();
let mut text = d.create_text().unwrap();
let mut map = d.create_map().unwrap();
batch_commit(doc.clone(), |_| {
arr.insert(0, Value::from(text.clone())).unwrap();
arr.insert(1, Value::from(map.clone())).unwrap();
});
batch_commit(doc.clone(), |_| {
text.insert(0, "hello world").unwrap();
text.remove(5, 6).unwrap();
});
batch_commit(doc.clone(), |_| {
map.insert("key".into(), 123).unwrap();
});
batch_commit(doc.clone(), |_| {
map.remove("key");
});
batch_commit(doc.clone(), |_| {
arr.remove(0, 1).unwrap();
});
});
});
}
}
packages/common/y-octo/core/src/doc/codec/any.rs
+11 -36
View File
@@ -40,10 +40,10 @@ impl<R: CrdtReader> CrdtRead<R> for Any {
0 => Ok(Any::Undefined),
1 => Ok(Any::Null),
// in yjs implementation, flag 2 only save 32bit integer
2 => Ok(Any::Integer(reader.read_var_i32()?)), // Integer
2 => Ok(Any::Integer(reader.read_var_i32()?)), // Integer
3 => Ok(Any::Float32(reader.read_f32_be()?.into())), // Float32
4 => Ok(Any::Float64(reader.read_f64_be()?.into())), // Float64
5 => Ok(Any::BigInt64(reader.read_i64_be()?)), // BigInt64
5 => Ok(Any::BigInt64(reader.read_i64_be()?)), // BigInt64
6 => Ok(Any::False),
7 => Ok(Any::True),
8 => Ok(Any::String(reader.read_var_string()?)), // String
@@ -57,9 +57,7 @@ impl<R: CrdtReader> CrdtRead<R> for Any {
} // Object
10 => {
let len = reader.read_var_u64()?;
let any = (0..len)
.map(|_| Self::read(reader))
.collect::<Result<Vec<_>, _>>()?;
let any = (0..len).map(|_| Self::read(reader)).collect::<Result<Vec<_>, _>>()?;
Ok(Any::Array(any))
} // Array
@@ -250,11 +248,7 @@ impl From<f64> for Any {
impl From<bool> for Any {
fn from(value: bool) -> Self {
if value {
Self::True
} else {
Self::False
}
if value { Self::True } else { Self::False }
}
}
@@ -344,11 +338,7 @@ impl From<&[u8]> for Any {
// TODO: impl for Any::Undefined
impl<T: Into<Any>> From<Option<T>> for Any {
fn from(value: Option<T>) -> Self {
if let Some(val) = value {
val.into()
} else {
Any::Null
}
if let Some(val) = value { val.into() } else { Any::Null }
}
}
@@ -374,12 +364,8 @@ impl From<serde_json::Value> for Any {
}
}
serde_json::Value::String(s) => Self::String(s),
serde_json::Value::Array(vec) => {
Self::Array(vec.into_iter().map(|v| v.into()).collect::<Vec<_>>())
}
serde_json::Value::Object(obj) => {
Self::Object(obj.into_iter().map(|(k, v)| (k, v.into())).collect())
}
serde_json::Value::Array(vec) => Self::Array(vec.into_iter().map(|v| v.into()).collect::<Vec<_>>()),
serde_json::Value::Object(obj) => Self::Object(obj.into_iter().map(|(k, v)| (k, v.into())).collect()),
}
}
}
@@ -587,10 +573,7 @@ mod tests {
Any::Object(
vec![
("type".to_string(), Any::String("Email".to_string())),
(
"address".to_string(),
Any::String("alice@example.com".to_string()),
),
("address".to_string(), Any::String("alice@example.com".to_string())),
]
.into_iter()
.collect(),
@@ -698,19 +681,11 @@ mod tests {
);
assert_eq!(
vec![("key".to_string(), 10u64.into())]
.into_iter()
.collect::<Any>(),
Any::Object(HashMap::from_iter(vec![(
"key".to_string(),
Any::Integer(10)
)]))
vec![("key".to_string(), 10u64.into())].into_iter().collect::<Any>(),
Any::Object(HashMap::from_iter(vec![("key".to_string(), Any::Integer(10))]))
);
let any: Any = 10u64.into();
assert_eq!(
[any].iter().collect::<Any>(),
Any::Array(vec![Any::Integer(10)])
);
assert_eq!([any].iter().collect::<Any>(), Any::Array(vec![Any::Integer(10)]));
}
}
packages/common/y-octo/core/src/doc/codec/content.rs
+15 -57
View File
@@ -90,16 +90,9 @@ impl std::fmt::Debug for Content {
.field("key", key)
.field("value", value)
.finish(),
Self::Type(arg0) => f
.debug_tuple("Type")
.field(&arg0.ty().unwrap().kind())
.finish(),
Self::Type(arg0) => f.debug_tuple("Type").field(&arg0.ty().unwrap().kind()).finish(),
Self::Any(arg0) => f.debug_tuple("Any").field(arg0).finish(),
Self::Doc { guid, opts } => f
.debug_struct("Doc")
.field("guid", guid)
.field("opts", opts)
.finish(),
Self::Doc { guid, opts } => f.debug_struct("Doc").field("guid", guid).field("opts", opts).finish(),
}
}
}
@@ -111,11 +104,7 @@ impl Content {
2 => {
let len = decoder.read_var_u64()?;
let strings = (0..len)
.map(|_| {
decoder
.read_var_string()
.map(|s| (s != "undefined").then_some(s))
})
.map(|_| decoder.read_var_string().map(|s| (s != "undefined").then_some(s)))
.collect::<Result<Vec<_>, _>>()?;
Ok(Self::Json(strings))
@@ -124,16 +113,14 @@ impl Content {
4 => Ok(Self::String(decoder.read_var_string()?)), // String
5 => {
let string = decoder.read_var_string()?;
let json =
serde_json::from_str(&string).map_err(|_| JwstCodecError::DamagedDocumentJson)?;
let json = serde_json::from_str(&string).map_err(|_| JwstCodecError::DamagedDocumentJson)?;
Ok(Self::Embed(json))
} // Embed
6 => {
let key = decoder.read_var_string()?;
let value = decoder.read_var_string()?;
let value =
serde_json::from_str(&value).map_err(|_| JwstCodecError::DamagedDocumentJson)?;
let value = serde_json::from_str(&value).map_err(|_| JwstCodecError::DamagedDocumentJson)?;
Ok(Self::Format { key, value })
} // Format
@@ -199,15 +186,11 @@ impl Content {
encoder.write_var_string(string)?;
}
Self::Embed(val) => {
encoder.write_var_string(
serde_json::to_string(val).map_err(|_| JwstCodecError::DamagedDocumentJson)?,
)?;
encoder.write_var_string(serde_json::to_string(val).map_err(|_| JwstCodecError::DamagedDocumentJson)?)?;
}
Self::Format { key, value } => {
encoder.write_var_string(key)?;
encoder.write_var_string(
serde_json::to_string(value).map_err(|_| JwstCodecError::DamagedDocumentJson)?,
)?;
encoder.write_var_string(serde_json::to_string(value).map_err(|_| JwstCodecError::DamagedDocumentJson)?)?;
}
Self::Type(ty) => {
if let Some(ty) = ty.ty() {
@@ -237,9 +220,7 @@ impl Content {
// TODO: need a custom wrapper with length cached, this cost too much
Self::String(string) => string.chars().map(|c| c.len_utf16()).sum::<usize>() as u64,
Self::Any(any) => any.len() as u64,
Self::Binary(_) | Self::Embed(_) | Self::Format { .. } | Self::Type(_) | Self::Doc { .. } => {
1
}
Self::Binary(_) | Self::Embed(_) | Self::Format { .. } | Self::Type(_) | Self::Doc { .. } => 1,
}
}
@@ -249,20 +230,14 @@ impl Content {
#[allow(dead_code)]
pub fn splittable(&self) -> bool {
matches!(
self,
Self::String { .. } | Self::Any { .. } | Self::Json { .. }
)
matches!(self, Self::String { .. } | Self::Any { .. } | Self::Json { .. })
}
pub fn split(&self, diff: u64) -> JwstCodecResult<(Self, Self)> {
match self {
Self::String(str) => {
let (left, right) = Self::split_as_utf16_str(str.as_str(), diff);
Ok((
Self::String(left.to_string()),
Self::String(right.to_string()),
))
Ok((Self::String(left.to_string()), Self::String(right.to_string())))
}
Self::Json(vec) => {
let (left, right) = vec.split_at(diff as usize);
@@ -321,11 +296,7 @@ mod tests {
loom_model!({
let contents = [
Content::Deleted(42),
Content::Json(vec![
None,
Some("test_1".to_string()),
Some("test_2".to_string()),
]),
Content::Json(vec![None, Some("test_1".to_string()), Some("test_2".to_string())]),
Content::Binary(vec![1, 2, 3]),
Content::String("hello".to_string()),
Content::Embed(Any::True),
@@ -336,10 +307,7 @@ mod tests {
Content::Type(YTypeRef::new(YTypeKind::Array, None)),
Content::Type(YTypeRef::new(YTypeKind::Map, None)),
Content::Type(YTypeRef::new(YTypeKind::Text, None)),
Content::Type(YTypeRef::new(
YTypeKind::XMLElement,
Some("test".to_string()),
)),
Content::Type(YTypeRef::new(YTypeKind::XMLElement, Some("test".to_string()))),
Content::Type(YTypeRef::new(YTypeKind::XMLFragment, None)),
Content::Type(YTypeRef::new(YTypeKind::XMLHook, Some("test".to_string()))),
Content::Type(YTypeRef::new(YTypeKind::XMLText, None)),
@@ -360,11 +328,7 @@ mod tests {
fn test_content_split() {
let contents = [
Content::String("hello".to_string()),
Content::Json(vec![
None,
Some("test_1".to_string()),
Some("test_2".to_string()),
]),
Content::Json(vec![None, Some("test_1".to_string()), Some("test_2".to_string())]),
Content::Any(vec![Any::BigInt64(42), Any::String("Test Any".to_string())]),
Content::Binary(vec![]),
];
@@ -390,18 +354,12 @@ mod tests {
let (left, right) = contents[2].split(1).unwrap();
assert!(contents[2].splittable());
assert_eq!(left, Content::Any(vec![Any::BigInt64(42)]));
assert_eq!(
right,
Content::Any(vec![Any::String("Test Any".to_string())])
);
assert_eq!(right, Content::Any(vec![Any::String("Test Any".to_string())]));
}
{
assert!(!contents[3].splittable());
assert_eq!(
contents[3].split(2),
Err(JwstCodecError::ContentSplitNotSupport(2))
);
assert_eq!(contents[3].split(2), Err(JwstCodecError::ContentSplitNotSupport(2)));
}
}
packages/common/y-octo/core/src/doc/codec/delete_set.rs
+3 -9
View File
@@ -1,5 +1,5 @@
use std::{
collections::{hash_map::Entry, VecDeque},
collections::{VecDeque, hash_map::Entry},
ops::{Deref, DerefMut, Range},
};
@@ -188,10 +188,7 @@ mod tests {
{
let mut delete_set = delete_set;
delete_set.add(1, 5, 10);
assert_eq!(
delete_set.get(&1),
Some(&OrderRange::from(vec![0..15, 20..30]))
);
assert_eq!(delete_set.get(&1), Some(&OrderRange::from(vec![0..15, 20..30])));
}
}
@@ -213,10 +210,7 @@ mod tests {
{
let mut delete_set = delete_set;
delete_set.batch_add_ranges(1, vec![40..50, 10..20]);
assert_eq!(
delete_set.get(&1),
Some(&OrderRange::from(vec![0..30, 40..50]))
);
assert_eq!(delete_set.get(&1), Some(&OrderRange::from(vec![0..30, 40..50])));
}
}
packages/common/y-octo/core/src/doc/codec/io/codec_v1.rs
+8 -33
View File
@@ -48,11 +48,7 @@ impl<'b> RawDecoder<'b> {
let pos = self.buffer.position() as usize;
let buf = self.buffer.into_inner();
if pos == 0 {
buf
} else {
&buf[pos..]
}
if pos == 0 { buf } else { &buf[pos..] }
}
}
@@ -88,24 +84,15 @@ impl CrdtReader for RawDecoder<'_> {
}
fn read_f32_be(&mut self) -> JwstCodecResult<f32> {
self
.buffer
.read_f32::<BigEndian>()
.map_err(reader::map_read_error)
self.buffer.read_f32::<BigEndian>().map_err(reader::map_read_error)
}
fn read_f64_be(&mut self) -> JwstCodecResult<f64> {
self
.buffer
.read_f64::<BigEndian>()
.map_err(reader::map_read_error)
self.buffer.read_f64::<BigEndian>().map_err(reader::map_read_error)
}
fn read_i64_be(&mut self) -> JwstCodecResult<i64> {
self
.buffer
.read_i64::<BigEndian>()
.map_err(reader::map_read_error)
self.buffer.read_i64::<BigEndian>().map_err(reader::map_read_error)
}
#[inline(always)]
@@ -151,22 +138,13 @@ impl CrdtWriter for RawEncoder {
Ok(())
}
fn write_f32_be(&mut self, num: f32) -> JwstCodecResult {
self
.buffer
.write_f32::<BigEndian>(num)
.map_err(writer::map_write_error)
self.buffer.write_f32::<BigEndian>(num).map_err(writer::map_write_error)
}
fn write_f64_be(&mut self, num: f64) -> JwstCodecResult {
self
.buffer
.write_f64::<BigEndian>(num)
.map_err(writer::map_write_error)
self.buffer.write_f64::<BigEndian>(num).map_err(writer::map_write_error)
}
fn write_i64_be(&mut self, num: i64) -> JwstCodecResult {
self
.buffer
.write_i64::<BigEndian>(num)
.map_err(writer::map_write_error)
self.buffer.write_i64::<BigEndian>(num).map_err(writer::map_write_error)
}
#[inline(always)]
@@ -197,10 +175,7 @@ mod tests {
let mut reader = RawDecoder::new(&[0x5, b'h', b'e', b'l', b'l', b'o']);
assert_eq!(reader.clone().read_var_string().unwrap(), "hello");
assert_eq!(
reader.clone().read_var_buffer().unwrap().as_slice(),
b"hello"
);
assert_eq!(reader.clone().read_var_buffer().unwrap().as_slice(), b"hello");
assert_eq!(reader.read_u8().unwrap(), 5);
assert_eq!(reader.read_u8().unwrap(), b'h');
packages/common/y-octo/core/src/doc/codec/item.rs
+17 -6
View File
@@ -232,12 +232,7 @@ impl Item {
!has_id && self.parent.is_some() || has_id && self.parent.is_none() && self.parent_sub.is_none()
}
pub fn read<R: CrdtReader>(
decoder: &mut R,
id: Id,
info: u8,
first_5_bit: u8,
) -> JwstCodecResult<Self> {
pub fn read<R: CrdtReader>(decoder: &mut R, id: Id, info: u8, first_5_bit: u8) -> JwstCodecResult<Self> {
let flags: ItemFlag = info.into();
let has_left_id = flags.check(item_flags::ITEM_HAS_LEFT_ID);
let has_right_id = flags.check(item_flags::ITEM_HAS_RIGHT_ID);
@@ -350,6 +345,22 @@ impl Item {
Ok(())
}
pub fn deep_compare(&self, other: &Self) -> bool {
if self.id != other.id
|| self.deleted() != other.deleted()
|| self.len() != other.len()
|| self.left.get().map(|l| l.last_id()) != other.left.get().map(|l| l.last_id())
|| self.right.get().map(|r| r.id) != other.right.get().map(|r| r.id)
|| self.origin_left_id != other.origin_left_id
|| self.origin_right_id != other.origin_right_id
|| self.parent_sub != other.parent_sub
{
return false;
}
true
}
}
#[allow(dead_code)]
packages/common/y-octo/core/src/doc/codec/mod.rs
+1 -1
View File
@@ -16,7 +16,7 @@ pub use delete_set::DeleteSet;
pub use id::{Client, Clock, Id};
pub use io::{CrdtRead, CrdtReader, CrdtWrite, CrdtWriter, RawDecoder, RawEncoder};
pub(crate) use item::{Item, ItemRef, Parent};
pub(crate) use item_flag::{item_flags, ItemFlag};
pub(crate) use item_flag::{ItemFlag, item_flags};
pub(crate) use refs::Node;
pub use update::Update;
#[cfg(test)]
packages/common/y-octo/core/src/doc/codec/refs.rs
+18 -20
View File
@@ -79,10 +79,10 @@ impl Node {
_ => {
let item = Somr::new(Item::read(decoder, id, info, first_5_bit)?);
if let Content::Type(ty) = &item.get().unwrap().content {
if let Some(mut ty) = ty.ty_mut() {
ty.item = item.clone();
}
if let Content::Type(ty) = &item.get().unwrap().content
&& let Some(mut ty) = ty.ty_mut()
{
ty.item = item.clone();
}
Ok(Node::Item(item))
@@ -282,8 +282,7 @@ impl Node {
l.extend(r.drain(0..));
}
(Content::String(l), Content::String(r)) => {
let allow_merge_string =
matches!(parent_kind, Some(YTypeKind::Text | YTypeKind::XMLText));
let allow_merge_string = matches!(parent_kind, Some(YTypeKind::Text | YTypeKind::XMLText));
if !allow_merge_string {
return false;
@@ -299,12 +298,11 @@ impl Node {
}
}
if let Some(Parent::Type(p)) = &litem.parent {
if let Some(parent) = p.ty_mut() {
if let Some(markers) = &parent.markers {
markers.replace_marker(rref.clone(), lref.clone(), -(llen as i64));
}
}
if let Some(Parent::Type(p)) = &litem.parent
&& let Some(parent) = p.ty_mut()
&& let Some(markers) = &parent.markers
{
markers.replace_marker(rref.clone(), lref.clone(), -(llen as i64));
}
if ritem.keep() {
@@ -455,15 +453,15 @@ mod tests {
#[cfg(not(loom))]
fn struct_info_round_trip(info: &mut Node) -> JwstCodecResult {
if let Node::Item(item) = info {
if let Some(item) = item.get_mut() {
if !item.is_valid() {
return Ok(());
}
if let Node::Item(item) = info
&& let Some(item) = item.get_mut()
{
if !item.is_valid() {
return Ok(());
}
if item.content.countable() {
item.flags.set_countable();
}
if item.content.countable() {
item.flags.set_countable();
}
}
let mut encoder = RawEncoder::default();
packages/common/y-octo/core/src/doc/codec/update.rs
+19 -50
View File
@@ -47,9 +47,7 @@ impl<R: CrdtReader> CrdtRead<R> for Update {
let delete_set = DeleteSet::read(decoder)?;
if !decoder.is_empty() {
return Err(JwstCodecError::UpdateNotFullyConsumed(
decoder.len() as usize
));
return Err(JwstCodecError::UpdateNotFullyConsumed(decoder.len() as usize));
}
Ok(Update {
@@ -282,23 +280,24 @@ impl<'a> UpdateIterator<'a> {
fn get_missing_dep(&self, struct_info: &Node) -> Option<Client> {
if let Some(item) = struct_info.as_item().get() {
let id = item.id;
if let Some(left) = &item.origin_left_id {
if left.client != id.client && left.clock >= self.state.get(&left.client) {
return Some(left.client);
}
if let Some(left) = &item.origin_left_id
&& left.client != id.client
&& left.clock >= self.state.get(&left.client)
{
return Some(left.client);
}
if let Some(right) = &item.origin_right_id {
if right.client != id.client && right.clock >= self.state.get(&right.client) {
return Some(right.client);
}
if let Some(right) = &item.origin_right_id
&& right.client != id.client
&& right.clock >= self.state.get(&right.client)
{
return Some(right.client);
}
if let Some(parent) = &item.parent {
match parent {
Parent::Id(parent_id)
if parent_id.client != id.client
&& parent_id.clock >= self.state.get(&parent_id.client) =>
if parent_id.client != id.client && parent_id.clock >= self.state.get(&parent_id.client) =>
{
return Some(parent_id.client);
}
@@ -319,15 +318,7 @@ impl<'a> UpdateIterator<'a> {
// Safety:
// client index of updates and update length are both checked in next_client
// safe to use unwrap
cur.replace(
self
.update
.structs
.get_mut(&client)
.unwrap()
.pop_front()
.unwrap(),
);
cur.replace(self.update.structs.get_mut(&client).unwrap().pop_front().unwrap());
}
cur
@@ -437,10 +428,7 @@ impl Iterator for DeleteSetIterator<'_> {
return Some((client, range));
} else {
// all state missing
self
.update
.pending_delete_set
.add(client, start, end - start);
self.update.pending_delete_set.add(client, start, end - start);
}
}
@@ -478,17 +466,9 @@ mod tests {
fn test_parse_doc() {
let docs = [
(include_bytes!("../../fixtures/basic.bin").to_vec(), 1, 188),
(
include_bytes!("../../fixtures/database.bin").to_vec(),
1,
149,
),
(include_bytes!("../../fixtures/database.bin").to_vec(), 1, 149),
(include_bytes!("../../fixtures/large.bin").to_vec(), 1, 9036),
(
include_bytes!("../../fixtures/with-subdoc.bin").to_vec(),
2,
30,
),
(include_bytes!("../../fixtures/with-subdoc.bin").to_vec(), 2, 30),
(
include_bytes!("../../fixtures/edge-case-left-right-same-node.bin").to_vec(),
2,
@@ -500,10 +480,7 @@ mod tests {
let update = parse_doc_update(doc).unwrap();
assert_eq!(update.structs.len(), clients);
assert_eq!(
update.structs.iter().map(|s| s.1.len()).sum::<usize>(),
structs
);
assert_eq!(update.structs.iter().map(|s| s.1.len()).sum::<usize>(), structs);
}
}
@@ -526,9 +503,7 @@ mod tests {
#[ignore = "just for local data test"]
#[test]
fn test_parse_local_doc() {
let json =
serde_json::from_slice::<Vec<Data>>(include_bytes!("../../fixtures/local_docs.json"))
.unwrap();
let json = serde_json::from_slice::<Vec<Data>>(include_bytes!("../../fixtures/local_docs.json")).unwrap();
for ws in json {
let data = &ws.blob[5..=(ws.blob.len() - 2)];
@@ -609,13 +584,7 @@ mod tests {
assert_eq!(iter.next(), None);
assert!(!update.pending_structs.is_empty());
assert_eq!(
update
.pending_structs
.get_mut(&0)
.unwrap()
.pop_front()
.unwrap()
.id(),
update.pending_structs.get_mut(&0).unwrap().pop_front().unwrap().id(),
(0, 4).into()
);
assert!(!update.missing_state.is_empty());
packages/common/y-octo/core/src/doc/codec/utils/items.rs
+2 -7
View File
@@ -7,9 +7,7 @@ pub(crate) struct ItemBuilder {
#[allow(dead_code)]
impl ItemBuilder {
pub fn new() -> ItemBuilder {
Self {
item: Item::default(),
}
Self { item: Item::default() }
}
pub fn id(mut self, id: Id) -> ItemBuilder {
@@ -93,10 +91,7 @@ mod tests {
assert_eq!(item.origin_right_id, Some(Id::new(4, 5)));
assert!(matches!(item.parent, Some(Parent::String(text)) if text == "test"));
assert_eq!(item.parent_sub, None);
assert_eq!(
item.content,
Content::Any(vec![Any::String("Hello".into())])
);
assert_eq!(item.content, Content::Any(vec![Any::String("Hello".into())]));
});
}
}
packages/common/y-octo/core/src/doc/common/range.rs
+13 -28
View File
@@ -73,10 +73,10 @@ impl OrderRange {
}
}
next_old = old_iter.next();
if let Some(next_old) = &next_old {
if next_old.start > new_range.end {
continue;
}
if let Some(next_old) = &next_old
&& next_old.start > new_range.end
{
continue;
}
}
next_new = new_iter.next();
@@ -184,10 +184,10 @@ impl OrderRange {
}
fn make_single(&mut self) {
if let OrderRange::Fragment(ranges) = self {
if ranges.len() == 1 {
*self = OrderRange::Range(ranges[0].clone());
}
if let OrderRange::Fragment(ranges) = self
&& ranges.len() == 1
{
*self = OrderRange::Range(ranges[0].clone());
}
}
@@ -278,10 +278,7 @@ impl<'a> IntoIterator for &'a OrderRange {
type IntoIter = OrderRangeIter<'a>;
fn into_iter(self) -> Self::IntoIter {
OrderRangeIter {
range: self,
idx: 0,
}
OrderRangeIter { range: self, idx: 0 }
}
}
@@ -394,18 +391,9 @@ mod tests {
assert!(OrderRange::check_range_covered(&[0..1], &[0..3]));
assert!(OrderRange::check_range_covered(&[1..2], &[0..3]));
assert!(OrderRange::check_range_covered(&[1..2, 2..3], &[0..3]));
assert!(!OrderRange::check_range_covered(
&[1..2, 2..3, 3..4],
&[0..3]
));
assert!(OrderRange::check_range_covered(
&[0..1, 2..3],
&[0..2, 2..4]
));
assert!(OrderRange::check_range_covered(
&[0..1, 2..3, 3..4],
&[0..2, 2..4]
),);
assert!(!OrderRange::check_range_covered(&[1..2, 2..3, 3..4], &[0..3]));
assert!(OrderRange::check_range_covered(&[0..1, 2..3], &[0..2, 2..4]));
assert!(OrderRange::check_range_covered(&[0..1, 2..3, 3..4], &[0..2, 2..4]),);
}
#[test]
@@ -469,10 +457,7 @@ mod tests {
fn iter() {
let range: OrderRange = vec![(0..10), (20..30)].into();
assert_eq!(
range.into_iter().collect::<Vec<_>>(),
vec![(0..10), (20..30)]
);
assert_eq!(range.into_iter().collect::<Vec<_>>(), vec![(0..10), (20..30)]);
let range: OrderRange = OrderRange::Range(0..10);
packages/common/y-octo/core/src/doc/common/somr.rs
+2 -6
View File
@@ -289,8 +289,7 @@ impl<T> FlattenGet<T> for Option<Somr<T>> {
impl<T: PartialEq> PartialEq for Somr<T> {
fn eq(&self, other: &Self) -> bool {
self.ptr() == other.ptr()
|| !self.dangling() && !other.dangling() && self.inner() == other.inner()
self.ptr() == other.ptr() || !self.dangling() && !other.dangling() && self.inner() == other.inner()
}
}
@@ -385,10 +384,7 @@ mod tests {
let five_ref = five.clone();
assert!(!five_ref.is_owned());
assert_eq!(five_ref.get(), Some(&5));
assert_eq!(
five_ref.ptr().as_ptr() as usize,
five.ptr().as_ptr() as usize
);
assert_eq!(five_ref.ptr().as_ptr() as usize, five.ptr().as_ptr() as usize);
drop(five);
// owner released
packages/common/y-octo/core/src/doc/common/state.rs
+2 -2
View File
@@ -1,8 +1,8 @@
use std::ops::{Deref, DerefMut};
use super::{
Client, ClientMap, Clock, CrdtRead, CrdtReader, CrdtWrite, CrdtWriter, HashMapExt, Id,
JwstCodecResult, HASHMAP_SAFE_CAPACITY,
Client, ClientMap, Clock, CrdtRead, CrdtReader, CrdtWrite, CrdtWriter, HASHMAP_SAFE_CAPACITY, HashMapExt, Id,
JwstCodecResult,
};
#[derive(Default, Debug, PartialEq, Clone)]
packages/common/y-octo/core/src/doc/document.rs
+78 -82
View File
@@ -1,7 +1,11 @@
#[cfg(feature = "events")]
use publisher::DocPublisher;
use super::{history::StoreHistory, store::StoreRef, *};
use super::{
history::StoreHistory,
store::{ChangedTypeRefs, StoreRef},
*,
};
use crate::sync::{Arc, RwLock};
#[cfg(feature = "debug")]
@@ -43,24 +47,6 @@ impl Default for DocOptions {
gc: true,
}
} else {
/// It tends to generate small numbers.
/// Since the client id will be included in all crdt items, the
/// small client helps to reduce the binary size.
///
/// NOTE: The probability of 36% of the random number generated by
/// this function is greater than [u32::MAX]
fn prefer_small_random() -> u64 {
use rand::{distr::Distribution, rng};
use rand_distr::Exp;
let scale_factor = u16::MAX as f64;
let v: f64 = Exp::new(1.0 / scale_factor)
.map(|exp| exp.sample(&mut rng()))
.unwrap_or_else(|_| rand::random());
(v * scale_factor) as u64
}
Self {
client_id: prefer_small_random(),
guid: nanoid::nanoid!(),
@@ -138,6 +124,7 @@ pub struct Doc {
pub(crate) store: StoreRef,
#[cfg(feature = "events")]
pub publisher: Arc<DocPublisher>,
pub(crate) batch: Somr<Batch>,
}
unsafe impl Send for Doc {}
@@ -171,6 +158,7 @@ impl Doc {
store,
#[cfg(feature = "events")]
publisher,
batch: Somr::none(),
}
}
@@ -182,6 +170,14 @@ impl Doc {
self.client_id
}
pub fn set_client(&mut self, client_id: u64) {
self.client_id = client_id;
}
pub fn renew_client(&mut self) {
self.client_id = prefer_small_random();
}
pub fn clients(&self) -> Vec<u64> {
self.store.read().unwrap().clients()
}
@@ -205,6 +201,17 @@ impl Doc {
}
}
pub(crate) fn get_changed(&self) -> ChangedTypeRefs {
self.store.write().unwrap().get_changed()
}
pub fn store_compare(&self, other: &Doc) -> bool {
let store = self.store.read().unwrap();
let other_store = other.store.read().unwrap();
store.deep_compare(&other_store)
}
pub fn options(&self) -> &DocOptions {
&self.opts
}
@@ -220,10 +227,7 @@ impl Doc {
Self::try_from_binary_v1_with_options(binary, DocOptions::default())
}
pub fn try_from_binary_v1_with_options<T: AsRef<[u8]>>(
binary: T,
options: DocOptions,
) -> JwstCodecResult<Self> {
pub fn try_from_binary_v1_with_options<T: AsRef<[u8]>>(binary: T, options: DocOptions) -> JwstCodecResult<Self> {
let mut doc = Doc::with_options(options);
doc.apply_update_from_binary_v1(binary)?;
Ok(doc)
@@ -316,9 +320,7 @@ impl Doc {
}
pub fn create_text(&self) -> JwstCodecResult<Text> {
YTypeBuilder::new(self.store.clone())
.with_kind(YTypeKind::Text)
.build()
YTypeBuilder::new(self.store.clone()).with_kind(YTypeKind::Text).build()
}
pub fn get_or_create_array<S: AsRef<str>>(&self, str: S) -> JwstCodecResult<Array> {
@@ -342,9 +344,7 @@ impl Doc {
}
pub fn create_map(&self) -> JwstCodecResult<Map> {
YTypeBuilder::new(self.store.clone())
.with_kind(YTypeKind::Map)
.build()
YTypeBuilder::new(self.store.clone()).with_kind(YTypeKind::Map).build()
}
pub fn get_map(&self, str: &str) -> JwstCodecResult<Map> {
@@ -378,6 +378,10 @@ impl Doc {
self.store.read().unwrap().get_state_vector()
}
pub fn get_delete_sets(&self) -> DeleteSet {
self.store.read().unwrap().get_delete_sets()
}
#[cfg(feature = "events")]
pub fn subscribe(&self, cb: impl Fn(&[u8], &[History]) + Sync + Send + 'static) {
self.publisher.subscribe(cb);
@@ -393,6 +397,11 @@ impl Doc {
self.publisher.count()
}
#[cfg(feature = "events")]
pub fn subscriber_count(&self) -> usize {
Arc::<DocPublisher>::strong_count(&self.publisher)
}
pub fn gc(&self) -> JwstCodecResult<()> {
self.store.write().unwrap().optimize()
}
@@ -400,7 +409,7 @@ impl Doc {
#[cfg(test)]
mod tests {
use yrs::{types::ToJson, updates::decoder::Decode, Array, Map, Options, Transact};
use yrs::{Array, Map, Options, Transact, types::ToJson, updates::decoder::Decode};
use super::*;
@@ -443,23 +452,14 @@ mod tests {
let mut doc = Doc::try_from_binary_v1(binary).unwrap();
let mut doc_new = Doc::try_from_binary_v1(binary_new).unwrap();
let diff_update = doc_new
.encode_state_as_update_v1(&doc.get_state_vector())
.unwrap();
let diff_update = doc_new.encode_state_as_update_v1(&doc.get_state_vector()).unwrap();
let diff_update_reverse = doc
.encode_state_as_update_v1(&doc_new.get_state_vector())
.unwrap();
let diff_update_reverse = doc.encode_state_as_update_v1(&doc_new.get_state_vector()).unwrap();
doc.apply_update_from_binary_v1(diff_update).unwrap();
doc_new
.apply_update_from_binary_v1(diff_update_reverse)
.unwrap();
doc_new.apply_update_from_binary_v1(diff_update_reverse).unwrap();
assert_eq!(
doc.encode_update_v1().unwrap(),
doc_new.encode_update_v1().unwrap()
);
assert_eq!(doc.encode_update_v1().unwrap(), doc_new.encode_update_v1().unwrap());
});
}
@@ -491,43 +491,43 @@ mod tests {
assert_json_diff::assert_json_matches!(array.to_json(&doc.transact()), json, config);
};
let binary = {
let doc = Doc::new();
let mut array = doc.get_or_create_array("abc").unwrap();
array.insert(0, 42).unwrap();
array.insert(1, -42).unwrap();
array.insert(2, true).unwrap();
array.insert(3, false).unwrap();
array.insert(4, "hello").unwrap();
array.insert(5, "world").unwrap();
{
let binary = {
let doc = Doc::new();
let mut array = doc.get_or_create_array("abc").unwrap();
array.insert(0, 42).unwrap();
array.insert(1, -42).unwrap();
array.insert(2, true).unwrap();
array.insert(3, false).unwrap();
array.insert(4, "hello").unwrap();
array.insert(5, "world").unwrap();
let mut sub_array = doc.create_array().unwrap();
array.insert(6, sub_array.clone()).unwrap();
// FIXME: array need insert first to compatible with yrs
sub_array.insert(0, 1).unwrap();
let mut sub_array = doc.create_array().unwrap();
array.insert(6, sub_array.clone()).unwrap();
// FIXME: array need insert first to compatible with yrs
sub_array.insert(0, 1).unwrap();
doc.encode_update_v1().unwrap()
};
doc.encode_update_v1().unwrap()
};
let ydoc = yrs::Doc::with_options(yrs_options);
let array = ydoc.get_or_insert_array("abc");
let mut trx = ydoc.transact_mut();
trx
.apply_update(yrs::Update::decode_v1(&binary).unwrap())
.unwrap();
let ydoc = yrs::Doc::with_options(yrs_options);
let array = ydoc.get_or_insert_array("abc");
let mut trx = ydoc.transact_mut();
trx.apply_update(yrs::Update::decode_v1(&binary).unwrap()).unwrap();
let config = assert_json_diff::Config::new(assert_json_diff::CompareMode::Strict)
.numeric_mode(assert_json_diff::NumericMode::AssumeFloat);
assert_json_diff::assert_json_matches!(array.to_json(&trx), json, config);
let config = assert_json_diff::Config::new(assert_json_diff::CompareMode::Strict)
.numeric_mode(assert_json_diff::NumericMode::AssumeFloat);
assert_json_diff::assert_json_matches!(array.to_json(&trx), json, config);
let mut doc = Doc::new();
let array = doc.get_or_create_array("abc").unwrap();
doc.apply_update_from_binary_v1(binary).unwrap();
let mut doc = Doc::new();
let array = doc.get_or_create_array("abc").unwrap();
doc.apply_update_from_binary_v1(binary).unwrap();
let list = array.iter().collect::<Vec<_>>();
let list = array.iter().collect::<Vec<_>>();
assert!(list.len() == 7);
assert!(matches!(list[6], Value::Array(_)));
assert!(list.len() == 7);
assert!(matches!(list[6], Value::Array(_)));
}
}
#[test]
@@ -551,11 +551,7 @@ mod tests {
count_clone2.fetch_add(1, Ordering::SeqCst);
});
doc_clone
.get_or_create_array("abc")
.unwrap()
.insert(0, 42)
.unwrap();
doc_clone.get_or_create_array("abc").unwrap().insert(0, 42).unwrap();
// wait observer, cycle once every 100mm
std::thread::sleep(std::time::Duration::from_millis(200));
@@ -594,8 +590,8 @@ mod tests {
doc
.apply_update_from_binary_v1(vec![
1, 1, 1, 1, 40, 0, 1, 0, 11, 115, 117, 98, 95, 109, 97, 112, 95, 107, 101, 121, 1, 119,
13, 115, 117, 98, 95, 109, 97, 112, 95, 118, 97, 108, 117, 101, 0,
1, 1, 1, 1, 40, 0, 1, 0, 11, 115, 117, 98, 95, 109, 97, 112, 95, 107, 101, 121, 1, 119, 13, 115, 117, 98, 95,
109, 97, 112, 95, 118, 97, 108, 117, 101, 0,
])
.unwrap();
@@ -612,8 +608,8 @@ mod tests {
.sum::<usize>();
doc
.apply_update_from_binary_v1(vec![
1, 1, 1, 1, 40, 0, 1, 0, 11, 115, 117, 98, 95, 109, 97, 112, 95, 107, 101, 121, 1, 119,
13, 115, 117, 98, 95, 109, 97, 112, 95, 118, 97, 108, 117, 101, 0,
1, 1, 1, 1, 40, 0, 1, 0, 11, 115, 117, 98, 95, 109, 97, 112, 95, 107, 101, 121, 1, 119, 13, 115, 117, 98, 95,
109, 97, 112, 95, 118, 97, 108, 117, 101, 0,
])
.unwrap();
packages/common/y-octo/core/src/doc/history.rs
+7 -18
View File
@@ -69,11 +69,7 @@ impl StoreHistory {
self.parse_items(store_items)
}
pub fn parse_delete_sets(
&self,
old_sets: &ClientMap<OrderRange>,
new_sets: &ClientMap<OrderRange>,
) -> Vec<History> {
pub fn parse_delete_sets(&self, old_sets: &ClientMap<OrderRange>, new_sets: &ClientMap<OrderRange>) -> Vec<History> {
let store = self.store.read().unwrap();
let deleted_items = new_sets
.iter()
@@ -109,11 +105,7 @@ impl StoreHistory {
let store = self.store.read().unwrap();
let mut sort_iter: Box<dyn Iterator<Item = Item>> = Box::new(
SortedNodes::new(if let Some(client) = client {
store
.items
.get(client)
.map(|i| vec![(client, i)])
.unwrap_or_default()
store.items.get(client).map(|i| vec![(client, i)]).unwrap_or_default()
} else {
store.items.iter().collect::<Vec<_>>()
})
@@ -273,10 +265,10 @@ impl Iterator for SortedNodes<'_> {
type Item = Node;
fn next(&mut self) -> Option<Self::Item> {
if let Some(current) = self.current.as_mut() {
if let Some(node) = current.pop_back() {
return Some(node);
}
if let Some(current) = self.current.as_mut()
&& let Some(node) = current.pop_back()
{
return Some(node);
}
if let Some((_, nodes)) = self.nodes.pop() {
@@ -318,10 +310,7 @@ mod test {
let update = doc.encode_update().unwrap();
assert_eq!(
history.parse_store(Default::default()),
history.parse_update(&update,)
);
assert_eq!(history.parse_store(Default::default()), history.parse_update(&update,));
});
}
}
packages/common/y-octo/core/src/doc/mod.rs
+2
View File
@@ -1,4 +1,5 @@
mod awareness;
mod batch;
mod codec;
mod common;
mod document;
@@ -12,6 +13,7 @@ mod utils;
pub use ahash::{HashMap, HashMapExt, HashSet, HashSetExt};
pub use awareness::{Awareness, AwarenessEvent};
pub use batch::{Batch, batch_commit};
pub use codec::*;
pub use common::*;
pub use document::{Doc, DocOptions};
packages/common/y-octo/core/src/doc/publisher.rs
+7 -12
View File
@@ -34,7 +34,10 @@ impl DocPublisher {
observing: Arc::new(AtomicBool::new(false)),
};
if cfg!(not(any(feature = "bench", fuzzing, loom, miri))) {
if cfg!(all(
feature = "subscribe",
not(any(feature = "bench", fuzzing, loom, miri))
)) {
publisher.start();
}
@@ -111,7 +114,7 @@ impl DocPublisher {
last_deletes = deletes;
for cb in subscribers.iter() {
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::panic::{AssertUnwindSafe, catch_unwind};
// catch panic if callback throw
catch_unwind(AssertUnwindSafe(|| {
cb(&binary, &history);
@@ -177,10 +180,7 @@ mod tests {
let ret = [
vec![vec!["(1, 0)", "test.key1", "val1"]],
vec![
vec!["(1, 1)", "test.key2", "val2"],
vec!["(1, 2)", "test.key3", "val3"],
],
vec![vec!["(1, 1)", "test.key2", "val2"], vec!["(1, 2)", "test.key3", "val3"]],
vec![
vec!["(1, 3)", "array.0", "val1"],
vec!["(1, 4)", "array.1", "val2"],
@@ -205,12 +205,7 @@ mod tests {
let ret = ret[cycle].clone();
for (i, h) in history.iter().enumerate() {
println!(
"history change by {} at {}: {}",
h.id,
h.parent.join("."),
h.content
);
println!("history change by {} at {}: {}", h.id, h.parent.join("."), h.content);
// lost first update by unknown reason in asan test, skip it if asan enabled
if option_env!("ASAN_OPTIONS").is_none() {
let ret = &ret[i];
packages/common/y-octo/core/src/doc/store.rs
+223 -175
View File
@@ -1,5 +1,5 @@
use std::{
collections::{hash_map::Entry, VecDeque},
collections::{VecDeque, hash_map::Entry},
mem,
ops::{Deref, Range},
};
@@ -10,6 +10,8 @@ use crate::{
sync::{Arc, RwLock, RwLockWriteGuard, Weak},
};
pub type ChangedTypeRefs = HashMap<YTypeRef, Vec<SmolStr>>;
unsafe impl Send for DocStore {}
unsafe impl Sync for DocStore {}
@@ -26,6 +28,8 @@ pub(crate) struct DocStore {
pub dangling_types: HashMap<usize, YTypeRef>,
pub pending: Option<Update>,
pub last_optimized_state: StateVector,
// changed item's parent, value is the parent's sub key if exists
pub changed: ChangedTypeRefs,
}
pub(crate) type StoreRef = Arc<RwLock<DocStore>>;
@@ -102,6 +106,10 @@ impl DocStore {
Self::items_as_state_vector(&self.items)
}
pub fn get_delete_sets(&self) -> DeleteSet {
self.delete_set.clone()
}
fn items_as_state_vector(items: &ClientMap<VecDeque<Node>>) -> StateVector {
let mut state = StateVector::default();
for (client, structs) in items.iter() {
@@ -175,10 +183,10 @@ impl DocStore {
let id = (self.client(), self.get_state(self.client())).into();
let item = Somr::new(Item::new(id, content, left, right, parent, parent_sub));
if let Content::Type(ty) = &item.get().unwrap().content {
if let Some(mut ty) = ty.ty_mut() {
ty.item = item.clone();
}
if let Content::Type(ty) = &item.get().unwrap().content
&& let Some(mut ty) = ty.ty_mut()
{
ty.item = item.clone();
}
item
@@ -190,10 +198,10 @@ impl DocStore {
pub fn get_node_with_idx<I: Into<Id>>(&self, id: I) -> Option<(Node, usize)> {
let id = id.into();
if let Some(items) = self.items.get(&id.client) {
if let Some(index) = Self::get_node_index(items, id.clock) {
return items.get(index).map(|item| (item.clone(), index));
}
if let Some(items) = self.items.get(&id.client)
&& let Some(index) = Self::get_node_index(items, id.clock)
{
return items.get(index).map(|item| (item.clone(), index));
}
None
@@ -204,20 +212,16 @@ impl DocStore {
let id = id.into();
if let Some(items) = self.items.get_mut(&id.client) {
if let Some(idx) = Self::get_node_index(items, id.clock) {
return Self::split_node_at(items, idx, diff);
}
if let Some(items) = self.items.get_mut(&id.client)
&& let Some(idx) = Self::get_node_index(items, id.clock)
{
return Self::split_node_at(items, idx, diff);
}
Err(JwstCodecError::StructSequenceNotExists(id.client))
}
pub fn split_node_at(
items: &mut VecDeque<Node>,
idx: usize,
diff: u64,
) -> JwstCodecResult<(Node, Node)> {
pub fn split_node_at(items: &mut VecDeque<Node>, idx: usize, diff: u64) -> JwstCodecResult<(Node, Node)> {
debug_assert!(diff > 0);
let node = items.get(idx).unwrap().clone();
@@ -263,16 +267,16 @@ impl DocStore {
pub fn split_at_and_get_right<I: Into<Id>>(&mut self, id: I) -> JwstCodecResult<Node> {
let id = id.into();
if let Some(items) = self.items.get_mut(&id.client) {
if let Some(index) = Self::get_node_index(items, id.clock) {
let item = items.get(index).unwrap().clone();
let offset = id.clock - item.clock();
if offset > 0 && item.is_item() {
let (_, right) = Self::split_node_at(items, index, offset)?;
return Ok(right);
} else {
return Ok(item);
}
if let Some(items) = self.items.get_mut(&id.client)
&& let Some(index) = Self::get_node_index(items, id.clock)
{
let item = items.get(index).unwrap().clone();
let offset = id.clock - item.clock();
if offset > 0 && item.is_item() {
let (_, right) = Self::split_node_at(items, index, offset)?;
return Ok(right);
} else {
return Ok(item);
}
}
@@ -281,16 +285,16 @@ impl DocStore {
pub fn split_at_and_get_left<I: Into<Id>>(&mut self, id: I) -> JwstCodecResult<Node> {
let id = id.into();
if let Some(items) = self.items.get_mut(&id.client) {
if let Some(index) = Self::get_node_index(items, id.clock) {
let item = items.get(index).unwrap().clone();
let offset = id.clock - item.clock();
if offset != item.len() - 1 && !item.is_gc() {
let (left, _) = Self::split_node_at(items, index, offset + 1)?;
return Ok(left);
} else {
return Ok(item);
}
if let Some(items) = self.items.get_mut(&id.client)
&& let Some(index) = Self::get_node_index(items, id.clock)
{
let item = items.get(index).unwrap().clone();
let offset = id.clock - item.clock();
if offset != item.len() - 1 && !item.is_gc() {
let (left, _) = Self::split_node_at(items, index, offset + 1)?;
return Ok(left);
} else {
return Ok(item);
}
}
@@ -430,12 +434,7 @@ impl DocStore {
Ok(())
}
pub fn integrate(
&mut self,
mut node: Node,
offset: u64,
parent: Option<&mut YType>,
) -> JwstCodecResult {
pub fn integrate(&mut self, mut node: Node, offset: u64, parent: Option<&mut YType>) -> JwstCodecResult {
match &mut node {
Node::Item(item_owner_ref) => {
assert!(
@@ -451,9 +450,7 @@ impl DocStore {
if offset > 0 {
this.id.clock += offset;
if let Node::Item(left_ref) =
self.split_at_and_get_left(Id::new(this.id.client, this.id.clock - 1))?
{
if let Node::Item(left_ref) = self.split_at_and_get_left(Id::new(this.id.client, this.id.clock - 1))? {
this.origin_left_id = left_ref.get().map(|left| left.last_id());
this.left = left_ref;
}
@@ -550,11 +547,7 @@ impl DocStore {
} else {
// no left, parent.start = this
right = if let Some(parent_sub) = &this.parent_sub {
parent
.map
.get(parent_sub)
.map(|n| Node::Item(n.clone()).head())
.into()
parent.map.get(parent_sub).map(|n| Node::Item(n.clone()).head()).into()
} else {
mem::replace(&mut parent.start, item_owner_ref.clone())
};
@@ -571,9 +564,7 @@ impl DocStore {
} else {
// no right, parent.start = this, delete this.left
if let Some(parent_sub) = &this.parent_sub {
parent
.map
.insert(parent_sub.clone(), item_owner_ref.clone());
parent.map.insert(parent_sub.clone(), item_owner_ref.clone());
if let Some(left) = this.left.get() {
self.delete_item(left, Some(parent));
@@ -582,11 +573,7 @@ impl DocStore {
}
this.right = right.clone();
let parent_deleted = parent
.item
.get()
.map(|item| item.deleted())
.unwrap_or(false);
let parent_deleted = parent.item.get().map(|item| item.deleted()).unwrap_or(false);
// should delete
if parent_deleted || this.parent_sub.is_some() && this.right.is_some() {
@@ -599,6 +586,9 @@ impl DocStore {
}
parent_lock.take();
// mark changed item's parent
Self::mark_changed(&mut self.changed, ty.clone(), this.parent_sub.clone());
} else {
// if parent not exists, integrate GC node instead
// don't delete it because it may referenced by other nodes
@@ -621,7 +611,7 @@ impl DocStore {
pub fn delete_item(&mut self, item: &Item, parent: Option<&mut YType>) {
let mut pending_delete_sets = HashMap::new();
Self::delete_item_inner(&mut pending_delete_sets, item, parent);
Self::delete_item_inner(&mut pending_delete_sets, &mut self.changed, item, parent);
for (client, ranges) in pending_delete_sets {
self.delete_set.batch_add_ranges(client, ranges);
}
@@ -629,6 +619,7 @@ impl DocStore {
fn delete_item_inner(
delete_set: &mut HashMap<u64, Vec<Range<u64>>>,
changed: &mut ChangedTypeRefs,
item: &Item,
parent: Option<&mut YType>,
) {
@@ -663,7 +654,7 @@ impl DocStore {
let mut item_ref = ty.start.clone();
while let Some(item) = item_ref.get() {
if !item.deleted() {
Self::delete_item_inner(delete_set, item, Some(&mut ty));
Self::delete_item_inner(delete_set, changed, item, Some(&mut ty));
}
item_ref = item.right.clone();
@@ -671,10 +662,10 @@ impl DocStore {
let map_values = ty.map.values().cloned().collect::<Vec<_>>();
for item in map_values {
if let Some(item) = item.get() {
if !item.deleted() {
Self::delete_item_inner(delete_set, item, Some(&mut ty));
}
if let Some(item) = item.get()
&& !item.deleted()
{
Self::delete_item_inner(delete_set, changed, item, Some(&mut ty));
}
}
}
@@ -684,6 +675,11 @@ impl DocStore {
}
_ => {}
}
// mark deleted item's parent
if let Some(Parent::Type(ty)) = &item.parent {
Self::mark_changed(changed, ty.clone(), item.parent_sub.clone());
}
}
pub fn delete_node(&mut self, struct_info: &Node, parent: Option<&mut YType>) {
@@ -696,59 +692,55 @@ impl DocStore {
let start = range.start;
let end = range.end;
if let Some(items) = self.items.get_mut(&client) {
if let Some(mut idx) = DocStore::get_node_index(items, start) {
{
// id.clock <= range.start < id.end
// need to split the item and delete the right part
// -----item-----
// ^start
let node = &items[idx];
let id = node.id();
if !node.deleted() && id.clock < start {
DocStore::split_node_at(items, idx, start - id.clock)?;
idx += 1;
}
};
let mut pending_delete_sets = HashMap::new();
while idx < items.len() {
let node = items[idx].clone();
let id = node.id();
if id.clock < end {
if !node.deleted() {
if let Some(item) = node.as_item().get() {
// need to split the item
// -----item-----
// ^end
if end < id.clock + node.len() {
DocStore::split_node_at(items, idx, end - id.clock)?;
}
Self::delete_item_inner(&mut pending_delete_sets, item, None);
}
}
} else {
break;
}
if let Some(items) = self.items.get_mut(&client)
&& let Some(mut idx) = DocStore::get_node_index(items, start)
{
{
// id.clock <= range.start < id.end
// need to split the item and delete the right part
// -----item-----
// ^start
let node = &items[idx];
let id = node.id();
if !node.deleted() && id.clock < start {
DocStore::split_node_at(items, idx, start - id.clock)?;
idx += 1;
}
for (client, ranges) in pending_delete_sets {
self.delete_set.batch_add_ranges(client, ranges);
};
let mut pending_delete_sets = HashMap::new();
while idx < items.len() {
let node = items[idx].clone();
let id = node.id();
if id.clock < end {
if !node.deleted()
&& let Some(item) = node.as_item().get()
{
// need to split the item
// -----item-----
// ^end
if end < id.clock + node.len() {
DocStore::split_node_at(items, idx, end - id.clock)?;
}
Self::delete_item_inner(&mut pending_delete_sets, &mut self.changed, item, None);
}
} else {
break;
}
idx += 1;
}
}
for (client, ranges) in pending_delete_sets {
self.delete_set.batch_add_ranges(client, ranges);
}
};
Ok(())
}
fn diff_state_vectors(
local_state_vector: &StateVector,
remote_state_vector: &StateVector,
) -> Vec<(Client, Clock)> {
fn diff_state_vectors(local_state_vector: &StateVector, remote_state_vector: &StateVector) -> Vec<(Client, Clock)> {
let mut diff = Vec::new();
for (client, &remote_clock) in remote_state_vector.iter() {
@@ -776,19 +768,28 @@ impl DocStore {
..Update::default()
};
if with_pending {
if let Some(pending) = &self.pending {
Update::merge_into(&mut update, [pending.clone()])
}
if with_pending && let Some(pending) = &self.pending {
Update::merge_into(&mut update, [pending.clone()])
}
Ok(update)
}
fn diff_structs(
map: &ClientMap<VecDeque<Node>>,
sv: &StateVector,
) -> JwstCodecResult<ClientMap<VecDeque<Node>>> {
fn mark_changed(changed: &mut ChangedTypeRefs, parent: YTypeRef, parent_sub: Option<SmolStr>) {
if parent.inner.is_some() {
let vec = changed.entry(parent).or_default();
if let Some(parent_sub) = parent_sub {
// only record the sub key if exists
vec.push(parent_sub);
}
}
}
pub fn get_changed(&mut self) -> ChangedTypeRefs {
mem::replace(&mut self.changed, HashMap::new())
}
fn diff_structs(map: &ClientMap<VecDeque<Node>>, sv: &StateVector) -> JwstCodecResult<ClientMap<VecDeque<Node>>> {
let local_state_vector = Self::items_as_state_vector(map);
let diff = Self::diff_state_vectors(&local_state_vector, sv);
let mut update_structs = ClientMap::new();
@@ -915,11 +916,11 @@ impl DocStore {
}
fn gc_content(content: &Content) -> JwstCodecResult {
if let Content::Type(ty) = content {
if let Some(mut ty) = ty.ty_mut() {
ty.start = Somr::none();
ty.map.clear();
}
if let Content::Type(ty) = content
&& let Some(mut ty) = ty.ty_mut()
{
ty.start = Somr::none();
ty.map.clear();
}
Ok(())
@@ -935,9 +936,7 @@ impl DocStore {
}
let nodes = self.items.get_mut(client).unwrap();
let first_change = Self::get_node_index(nodes, before_state)
.unwrap_or(1)
.max(1);
let first_change = Self::get_node_index(nodes, before_state).unwrap_or(1).max(1);
let mut idx = nodes.len() - 1;
while idx > 0 && idx >= first_change {
@@ -969,6 +968,39 @@ impl DocStore {
// return the index of processed items
idx - pos
}
pub fn deep_compare(&self, other: &Self) -> bool {
if self.items.len() != other.items.len() {
return false;
}
for (client, structs) in self.items.iter() {
if let Some(other_structs) = other.items.get(client) {
if structs.len() != other_structs.len() {
return false;
}
for (struct_info, other_struct_info) in structs.iter().zip(other_structs.iter()) {
if struct_info != other_struct_info {
return false;
}
if let (Node::Item(item), Node::Item(other_item)) = (struct_info, other_struct_info)
&& !match (item.get(), other_item.get()) {
(Some(item), Some(other_item)) => item.deep_compare(other_item),
(None, None) => true,
_ => false,
}
{
return false;
}
}
} else {
return false;
}
}
true
}
}
#[cfg(test)]
@@ -991,10 +1023,9 @@ mod tests {
let struct_info1 = Node::new_gc(Id::new(1, 1), 5);
let struct_info2 = Node::new_skip(Id::new(1, 6), 7);
doc_store.items.insert(
client_id,
VecDeque::from([struct_info1, struct_info2.clone()]),
);
doc_store
.items
.insert(client_id, VecDeque::from([struct_info1, struct_info2.clone()]));
let state = doc_store.get_state(client_id);
@@ -1022,24 +1053,15 @@ mod tests {
let struct_info2 = Node::new_gc((2, 0).into(), 6);
let struct_info3 = Node::new_skip((2, 6).into(), 1);
doc_store.items.insert(client1, VecDeque::from([struct_info1.clone()]));
doc_store
.items
.insert(client1, VecDeque::from([struct_info1.clone()]));
doc_store.items.insert(
client2,
VecDeque::from([struct_info2, struct_info3.clone()]),
);
.insert(client2, VecDeque::from([struct_info2, struct_info3.clone()]));
let state_map = doc_store.get_state_vector();
assert_eq!(
state_map.get(&client1),
struct_info1.clock() + struct_info1.len()
);
assert_eq!(
state_map.get(&client2),
struct_info3.clock() + struct_info3.len()
);
assert_eq!(state_map.get(&client1), struct_info1.clock() + struct_info1.len());
assert_eq!(state_map.get(&client2), struct_info3.clock() + struct_info3.len());
assert!(doc_store.self_check().is_ok());
});
@@ -1059,10 +1081,7 @@ mod tests {
assert!(doc_store.add_node(struct_info2).is_ok());
assert_eq!(
doc_store.add_node(struct_info3_err),
Err(JwstCodecError::StructClockInvalid {
expect: 6,
actually: 5
})
Err(JwstCodecError::StructClockInvalid { expect: 6, actually: 5 })
);
assert!(doc_store.add_node(struct_info3.clone()).is_ok());
assert_eq!(
@@ -1163,15 +1182,64 @@ mod tests {
// s1 used to be (1, 4), but it actually ref of first item in store, so now it
// should be (1, 2)
assert_eq!(
s1, left,
"doc internal mutation should not modify the pointer"
);
assert_eq!(s1, left, "doc internal mutation should not modify the pointer");
let right = doc_store.split_at_and_get_right((1, 5)).unwrap();
assert_eq!(right.len(), 3); // base => b_ase
});
}
#[test]
fn should_mark_changed_items() {
loom_model!({
let doc = DocOptions::new().with_client_id(1).build();
let mut arr = doc.get_or_create_array("arr").unwrap();
let mut text = doc.create_text().unwrap();
let mut map = doc.create_map().unwrap();
arr.insert(0, Value::from(text.clone())).unwrap();
arr.insert(1, Value::from(map.clone())).unwrap();
{
let changed = doc.store.write().unwrap().get_changed();
// for array, we will only record the type ref itself
assert_eq!(changed.len(), 1);
assert_eq!(changed.get(&arr.0), Some(&vec![]));
}
text.insert(0, "hello world").unwrap();
text.remove(5, 6).unwrap();
{
let changed = doc.store.write().unwrap().get_changed();
assert_eq!(changed.len(), 1);
assert_eq!(changed.get(&text.0), Some(&vec![]));
}
map.insert("key".into(), 123).unwrap();
{
let changed = doc.store.write().unwrap().get_changed();
assert_eq!(changed.len(), 1);
assert_eq!(changed.get(&map.0), Some(&vec!["key".into()]));
}
map.remove("key");
{
let changed = doc.store.write().unwrap().get_changed();
assert_eq!(changed.len(), 1);
assert_eq!(changed.get(&map.0), Some(&vec!["key".into()]));
}
arr.remove(0, 1).unwrap();
{
let changed = doc.store.write().unwrap().get_changed();
assert_eq!(changed.len(), 2);
// text's children mark parent(text) changed
assert_eq!(changed.get(&text.0), Some(&vec![]));
// text mark parent(arr) changed
assert_eq!(changed.get(&arr.0), Some(&vec![]));
}
});
}
#[test]
fn should_replace_gc_item_with_content_deleted() {
loom_model!({
@@ -1195,13 +1263,7 @@ mod tests {
store.gc_delete_set().unwrap();
assert_eq!(
&store
.get_node((1, 0))
.unwrap()
.as_item()
.get()
.unwrap()
.content,
&store.get_node((1, 0)).unwrap().as_item().get().unwrap().content,
&Content::Deleted(4)
);
});
@@ -1226,13 +1288,7 @@ mod tests {
assert_eq!(arr.len(), 0);
assert_eq!(
&store
.get_node((1, 0))
.unwrap()
.as_item()
.get()
.unwrap()
.content,
&store.get_node((1, 0)).unwrap().as_item().get().unwrap().content,
&Content::Deleted(1)
);
@@ -1256,9 +1312,7 @@ mod tests {
let mut pages = doc.get_or_create_map("pages").unwrap();
let page1 = doc.create_text().unwrap();
let mut page1_ref = page1.clone();
pages
.insert("page1".to_string(), Value::from(page1))
.unwrap();
pages.insert("page1".to_string(), Value::from(page1)).unwrap();
page1_ref.insert(0, "hello").unwrap();
doc.encode_update_v1().unwrap()
};
@@ -1276,13 +1330,7 @@ mod tests {
store.gc_delete_set().unwrap();
assert_eq!(
&store
.get_node((1, 0))
.unwrap()
.as_item()
.get()
.unwrap()
.content,
&store.get_node((1, 0)).unwrap().as_item().get().unwrap().content,
&Content::Deleted(1)
);
packages/common/y-octo/core/src/doc/types/array.rs
+29 -28
View File
@@ -52,6 +52,11 @@ impl Iterator for ArrayIter<'_> {
}
impl Array {
#[inline(always)]
pub fn id(&self) -> Option<Id> {
self._id()
}
#[inline]
pub fn len(&self) -> u64 {
self.content_len()
@@ -126,15 +131,26 @@ mod tests {
array.insert(0, "Hello").unwrap();
array.insert(2, "World").unwrap();
assert_eq!(
array.get(0).unwrap(),
Value::Any(Any::String("Hello".into()))
);
assert_eq!(array.get(0).unwrap(), Value::Any(Any::String("Hello".into())));
assert_eq!(array.get(1).unwrap(), Value::Any(Any::String(" ".into())));
assert_eq!(
array.get(2).unwrap(),
Value::Any(Any::String("World".into()))
);
assert_eq!(array.get(2).unwrap(), Value::Any(Any::String("World".into())));
});
}
#[test]
fn test_yarray_delete() {
let options = DocOptions::default();
loom_model!({
let doc = Doc::with_options(options.clone());
let mut array = doc.get_or_create_array("abc").unwrap();
array.insert(0, " ").unwrap();
array.insert(0, "Hello").unwrap();
array.insert(2, "World").unwrap();
array.remove(0, 2).unwrap();
assert_eq!(array.get(0).unwrap(), Value::Any(Any::String("World".into())));
});
}
@@ -163,15 +179,9 @@ mod tests {
doc.apply_update(update).unwrap();
let array = doc.get_or_create_array("abc").unwrap();
assert_eq!(
array.get(0).unwrap(),
Value::Any(Any::String("Hello".into()))
);
assert_eq!(array.get(0).unwrap(), Value::Any(Any::String("Hello".into())));
assert_eq!(array.get(5).unwrap(), Value::Any(Any::String(" ".into())));
assert_eq!(
array.get(6).unwrap(),
Value::Any(Any::String("World".into()))
);
assert_eq!(array.get(6).unwrap(), Value::Any(Any::String("World".into())));
assert_eq!(array.get(11).unwrap(), Value::Any(Any::String("!".into())));
});
@@ -196,15 +206,9 @@ mod tests {
doc.apply_update(update).unwrap();
let array = doc.get_or_create_array("abc").unwrap();
assert_eq!(
array.get(0).unwrap(),
Value::Any(Any::String("Hello".into()))
);
assert_eq!(array.get(0).unwrap(), Value::Any(Any::String("Hello".into())));
assert_eq!(array.get(5).unwrap(), Value::Any(Any::String(" ".into())));
assert_eq!(
array.get(6).unwrap(),
Value::Any(Any::String("World".into()))
);
assert_eq!(array.get(6).unwrap(), Value::Any(Any::String("World".into())));
assert_eq!(array.get(11).unwrap(), Value::Any(Any::String("!".into())));
});
}
@@ -237,10 +241,7 @@ mod tests {
.unwrap();
let arr = doc.get_or_create_array("abc").unwrap();
assert_eq!(
arr.get(2).unwrap(),
Value::Any(Any::String("world".to_string()))
)
assert_eq!(arr.get(2).unwrap(), Value::Any(Any::String("world".to_string())))
});
}
}
packages/common/y-octo/core/src/doc/types/list/mod.rs
+54 -44
View File
@@ -6,6 +6,7 @@ pub(crate) use search_marker::MarkerList;
use super::*;
#[derive(Debug)]
pub(crate) struct ItemPosition {
pub parent: YTypeRef,
pub left: ItemRef,
@@ -55,6 +56,11 @@ impl ItemPosition {
}
pub(crate) trait ListType: AsInner<Inner = YTypeRef> {
#[inline(always)]
fn _id(&self) -> Option<Id> {
self.as_inner().ty().and_then(|ty| ty.item.get().map(|item| item.id))
}
#[inline(always)]
fn content_len(&self) -> u64 {
self.as_inner().ty().unwrap().len
@@ -84,23 +90,29 @@ pub(crate) trait ListType: AsInner<Inner = YTypeRef> {
return Some(pos);
}
if let Some(markers) = &inner.markers {
if let Some(marker) = markers.find_marker(inner, index) {
if marker.index > remaining {
remaining = 0
} else {
remaining -= marker.index;
}
pos.index = marker.index;
pos.left = marker
.ptr
.get()
.map(|ptr| ptr.left.clone())
.unwrap_or_default();
pos.right = marker.ptr;
if let Some(markers) = &inner.markers
&& let Some(marker) = markers.find_marker(inner, index)
{
if marker.index > remaining {
remaining = 0
} else {
remaining -= marker.index;
}
pos.index = marker.index;
pos.left = marker.ptr.get().map(|ptr| ptr.left.clone()).unwrap_or_default();
pos.right = marker.ptr;
};
// avoid the first item of the list being deleted
while let Some(item) = pos.right.get() {
if item.deleted() {
pos.right = item.right.clone();
continue;
} else {
break;
}
}
while remaining > 0 {
if let Some(item) = pos.right.get() {
if item.indexable() {
@@ -141,16 +153,11 @@ pub(crate) trait ListType: AsInner<Inner = YTypeRef> {
Ok(())
}
fn insert_after(
ty: &mut YType,
store: &mut DocStore,
pos: ItemPosition,
content: Content,
) -> JwstCodecResult {
if let Some(markers) = &ty.markers {
if content.countable() {
markers.update_marker_changes(pos.index, content.clock_len() as i64);
}
fn insert_after(ty: &mut YType, store: &mut DocStore, pos: ItemPosition, content: Content) -> JwstCodecResult {
if let Some(markers) = &ty.markers
&& content.countable()
{
markers.update_marker_changes(pos.index, content.clock_len() as i64);
}
let item = store.create_item(
@@ -189,7 +196,12 @@ pub(crate) trait ListType: AsInner<Inner = YTypeRef> {
return Ok(());
}
if idx >= self.content_len() {
let content_len = self.content_len();
if content_len == 0 {
return Ok(());
}
if idx >= content_len {
return Err(JwstCodecError::IndexOutOfBound(idx));
}
@@ -204,34 +216,32 @@ pub(crate) trait ListType: AsInner<Inner = YTypeRef> {
Ok(())
}
fn remove_after(
ty: &mut YType,
store: &mut DocStore,
mut pos: ItemPosition,
len: u64,
) -> JwstCodecResult {
fn remove_after(ty: &mut YType, store: &mut DocStore, mut pos: ItemPosition, len: u64) -> JwstCodecResult {
pos.normalize(store)?;
let mut remaining = len;
while remaining > 0 {
if let Some(item) = pos.right.get() {
if item.indexable() {
let content_len = item.len();
if remaining < content_len {
store.split_node(item.id, remaining)?;
remaining = 0;
} else {
remaining -= content_len;
}
let item_ref = pos.right.clone();
let Some((indexable, content_len, item_id)) = item_ref.get().map(|item| (item.indexable(), item.len(), item.id))
else {
break;
};
store.delete_item(item, Some(ty));
if indexable {
if remaining < content_len {
store.split_node(item_id, remaining)?;
remaining = 0;
} else {
remaining -= content_len;
}
pos.forward();
} else {
break;
if let Some(item) = item_ref.get() {
store.delete_item(item, Some(ty));
}
}
pos.forward();
}
if let Some(markers) = &ty.markers {
packages/common/y-octo/core/src/doc/types/list/search_marker.rs
+3 -10
View File
@@ -69,11 +69,7 @@ impl MarkerList {
}
// mark pos and push to the end of the linked list
fn mark_position(
list: &mut VecDeque<SearchMarker>,
ptr: Somr<Item>,
index: u64,
) -> Option<SearchMarker> {
fn mark_position(list: &mut VecDeque<SearchMarker>, ptr: Somr<Item>, index: u64) -> Option<SearchMarker> {
if list.len() >= MAX_SEARCH_MARKER {
let mut oldest_marker = list.pop_front().unwrap();
oldest_marker.overwrite_marker(ptr, index);
@@ -126,9 +122,7 @@ impl MarkerList {
let mut list = self.borrow_mut();
let marker = list
.iter_mut()
.min_by_key(|m| (index as i64 - m.index as i64).abs());
let marker = list.iter_mut().min_by_key(|m| (index as i64 - m.index as i64).abs());
let mut marker_index = marker.as_ref().map(|m| m.index).unwrap_or(0);
@@ -201,8 +195,7 @@ impl MarkerList {
match marker {
Some(marker)
if (marker.index as f64 - marker_index as f64).abs()
< parent.len as f64 / MAX_SEARCH_MARKER as f64 =>
if (marker.index as f64 - marker_index as f64).abs() < parent.len as f64 / MAX_SEARCH_MARKER as f64 =>
{
// adjust existing marker
marker.overwrite_marker(item_ptr, marker_index);
packages/common/y-octo/core/src/doc/types/map.rs
+26 -32
View File
@@ -2,13 +2,18 @@ use std::{collections::hash_map::Iter, rc::Rc};
use super::*;
use crate::{
JwstCodecResult,
doc::{AsInner, Node, Parent, YTypeRef},
impl_type, JwstCodecResult,
impl_type,
};
impl_type!(Map);
pub(crate) trait MapType: AsInner<Inner = YTypeRef> {
fn _id(&self) -> Option<Id> {
self.as_inner().ty().and_then(|ty| ty.item.get().map(|item| item.id))
}
fn _insert<V: Into<Value>>(&mut self, key: String, value: V) -> JwstCodecResult {
if let Some((mut store, mut ty)) = self.as_inner().write() {
let left = ty.map.get(&SmolStr::new(&key)).cloned();
@@ -54,12 +59,11 @@ pub(crate) trait MapType: AsInner<Inner = YTypeRef> {
}
fn _remove(&mut self, key: &str) {
if let Some((mut store, mut ty)) = self.as_inner().write() {
if let Some(item) = ty.map.get(key).cloned() {
if let Some(item) = item.get() {
store.delete_item(item, Some(&mut ty));
}
}
if let Some((mut store, mut ty)) = self.as_inner().write()
&& let Some(item) = ty.map.get(key).cloned()
&& let Some(item) = item.get()
{
store.delete_item(item, Some(&mut ty));
}
}
@@ -113,10 +117,10 @@ impl<'a> Iterator for EntriesInnerIterator<'a> {
fn next(&mut self) -> Option<Self::Item> {
if let Some(iter) = &mut self.iter {
for (k, v) in iter {
if let Some(item) = v.get() {
if !item.deleted() {
return Some((k.as_str(), item));
}
if let Some(item) = v.get()
&& !item.deleted()
{
return Some((k.as_str(), item));
}
}
@@ -154,6 +158,11 @@ impl<'a> Iterator for EntriesIterator<'a> {
impl MapType for Map {}
impl Map {
#[inline(always)]
pub fn id(&self) -> Option<Id> {
self._id()
}
#[inline(always)]
pub fn insert<V: Into<Value>>(&mut self, key: String, value: V) -> JwstCodecResult {
self._insert(key, value)
@@ -220,7 +229,7 @@ impl serde::Serialize for Map {
#[cfg(test)]
mod tests {
use super::*;
use crate::{loom_model, Any, Doc};
use crate::{Any, Doc, loom_model};
#[test]
fn test_map_basic() {
@@ -228,10 +237,7 @@ mod tests {
let doc = Doc::new();
let mut map = doc.get_or_create_map("map").unwrap();
map.insert("1".to_string(), "value").unwrap();
assert_eq!(
map.get("1").unwrap(),
Value::Any(Any::String("value".to_string()))
);
assert_eq!(map.get("1").unwrap(), Value::Any(Any::String("value".to_string())));
assert!(!map.contains_key("nonexistent_key"));
assert_eq!(map.len(), 1);
assert!(map.contains_key("1"));
@@ -252,10 +258,7 @@ mod tests {
let binary = doc.encode_update_v1().unwrap();
let new_doc = Doc::try_from_binary_v1(binary).unwrap();
let map = new_doc.get_or_create_map("map").unwrap();
assert_eq!(
map.get("1").unwrap(),
Value::Any(Any::String("value".to_string()))
);
assert_eq!(map.get("1").unwrap(), Value::Any(Any::String("value".to_string())));
assert_eq!(map.get("2").unwrap(), Value::Any(Any::False));
assert_eq!(map.len(), 2);
});
@@ -268,10 +271,7 @@ mod tests {
let mut map = doc.get_or_create_map("map").unwrap();
map.insert("1".to_string(), "value").unwrap();
map.insert("1".to_string(), "value2").unwrap();
assert_eq!(
map.get("1").unwrap(),
Value::Any(Any::String("value2".to_string()))
);
assert_eq!(map.get("1").unwrap(), Value::Any(Any::String("value2".to_string())));
assert_eq!(map.len(), 1);
});
}
@@ -290,14 +290,8 @@ mod tests {
{
let doc = Doc::try_from_binary_v1(binary).unwrap();
let map = doc.get_or_create_map("map").unwrap();
assert_eq!(
map.get("1").unwrap(),
Value::Any(Any::String("value1".to_string()))
);
assert_eq!(
map.get("2").unwrap(),
Value::Any(Any::String("value2".to_string()))
);
assert_eq!(map.get("1").unwrap(), Value::Any(Any::String("value1".to_string())));
assert_eq!(map.get("2").unwrap(), Value::Any(Any::String("value2".to_string())));
}
});
}
packages/common/y-octo/core/src/doc/types/mod.rs
+20 -20
View File
@@ -5,7 +5,11 @@ mod text;
mod value;
mod xml;
use std::{collections::hash_map::Entry, sync::Weak};
use std::{
collections::hash_map::Entry,
hash::{Hash, Hasher},
sync::Weak,
};
pub use array::*;
use list::*;
@@ -19,8 +23,8 @@ use super::{
*,
};
use crate::{
sync::{Arc, RwLock, RwLockReadGuard, RwLockWriteGuard},
Item, JwstCodecError, JwstCodecResult,
sync::{Arc, RwLock, RwLockReadGuard, RwLockWriteGuard},
};
#[derive(Debug, Default)]
@@ -45,9 +49,7 @@ pub(crate) struct YTypeRef {
impl PartialEq for YType {
fn eq(&self, other: &Self) -> bool {
self.root_name == other.root_name
|| (self.start.is_some() && self.start == other.start)
|| self.map == other.map
self.root_name == other.root_name || (self.start.is_some() && self.start == other.start) || self.map == other.map
}
}
@@ -62,6 +64,14 @@ impl PartialEq for YTypeRef {
}
}
impl Eq for YTypeRef {}
impl Hash for YTypeRef {
fn hash<H: Hasher>(&self, state: &mut H) {
self.inner.ptr().hash(state);
}
}
impl YType {
pub fn new(kind: YTypeKind, tag_name: Option<String>) -> Self {
YType {
@@ -129,15 +139,11 @@ impl YTypeRef {
#[allow(dead_code)]
pub fn read(&self) -> Option<(RwLockReadGuard<'_, DocStore>, RwLockReadGuard<'_, YType>)> {
self
.store()
.and_then(|store| self.ty().map(|ty| (store, ty)))
self.store().and_then(|store| self.ty().map(|ty| (store, ty)))
}
pub fn write(&self) -> Option<(RwLockWriteGuard<'_, DocStore>, RwLockWriteGuard<'_, YType>)> {
self
.store_mut()
.and_then(|store| self.ty_mut().map(|ty| (store, ty)))
self.store_mut().and_then(|store| self.ty_mut().map(|ty| (store, ty)))
}
}
@@ -238,9 +244,7 @@ impl YTypeBuilder {
let ty_ref = ty.clone();
store
.dangling_types
.insert(ty.inner.ptr().as_ptr() as usize, ty);
store.dangling_types.insert(ty.inner.ptr().as_ptr() as usize, ty);
ty_ref
};
@@ -338,14 +342,10 @@ macro_rules! impl_type {
inner.set_kind(super::YTypeKind::$name)?;
Ok($name::new(value.clone()))
}
_ => Err($crate::JwstCodecError::TypeCastError(std::stringify!(
$name
))),
_ => Err($crate::JwstCodecError::TypeCastError(std::stringify!($name))),
}
} else {
Err($crate::JwstCodecError::TypeCastError(std::stringify!(
$name
)))
Err($crate::JwstCodecError::TypeCastError(std::stringify!($name)))
}
}
}
packages/common/y-octo/core/src/doc/types/text.rs
+40 -72
View File
@@ -1,9 +1,10 @@
use std::{collections::BTreeMap, fmt::Display};
use super::{list::ListType, AsInner};
use super::{AsInner, list::ListType};
use crate::{
Any, Content, JwstCodecError, JwstCodecResult,
doc::{DocStore, ItemRef, Node, Parent, Somr, YType, YTypeRef},
impl_type, Any, Content, JwstCodecError, JwstCodecResult,
impl_type,
};
impl_type!(Text);
@@ -85,21 +86,12 @@ impl Text {
Content::Json(values) => {
let converted = values
.iter()
.map(|value| {
value
.as_ref()
.map(|s| Any::String(s.clone()))
.unwrap_or(Any::Undefined)
})
.map(|value| value.as_ref().map(|s| Any::String(s.clone())).unwrap_or(Any::Undefined))
.collect::<Vec<_>>();
push_insert(&mut ops, TextInsert::Embed(converted), &attrs);
}
Content::Binary(value) => {
push_insert(
&mut ops,
TextInsert::Embed(vec![Any::Binary(value.clone())]),
&attrs,
);
push_insert(&mut ops, TextInsert::Embed(vec![Any::Binary(value.clone())]), &attrs);
}
_ => {}
}
@@ -121,13 +113,7 @@ impl Text {
let attrs = format.clone().unwrap_or_default();
match insert {
TextInsert::Text(text) => {
insert_text_content(
&mut store,
&mut ty,
&mut pos,
Content::String(text.clone()),
attrs,
)?;
insert_text_content(&mut store, &mut ty, &mut pos, Content::String(text.clone()), attrs)?;
}
TextInsert::Embed(values) => {
for value in values {
@@ -225,38 +211,29 @@ fn is_nullish(value: &Any) -> bool {
}
fn push_insert(ops: &mut Vec<TextDeltaOp>, insert: TextInsert, attrs: &TextAttributes) {
let format = if attrs.is_empty() {
None
} else {
Some(attrs.clone())
};
let format = if attrs.is_empty() { None } else { Some(attrs.clone()) };
if let Some(TextDeltaOp::Insert {
insert: TextInsert::Text(prev),
format: prev_format,
}) = ops.last_mut()
&& let TextInsert::Text(text) = insert
{
if let TextInsert::Text(text) = insert {
if prev_format.as_ref() == format.as_ref() {
prev.push_str(&text);
return;
}
ops.push(TextDeltaOp::Insert {
insert: TextInsert::Text(text),
format,
});
if prev_format.as_ref() == format.as_ref() {
prev.push_str(&text);
return;
}
ops.push(TextDeltaOp::Insert {
insert: TextInsert::Text(text),
format,
});
return;
}
ops.push(TextDeltaOp::Insert { insert, format });
}
fn advance_text_position(
store: &mut DocStore,
pos: &mut TextPosition,
mut remaining: u64,
) -> JwstCodecResult {
fn advance_text_position(store: &mut DocStore, pos: &mut TextPosition, mut remaining: u64) -> JwstCodecResult {
while remaining > 0 {
let Some(item) = pos.right.get() else {
return Err(JwstCodecError::IndexOutOfBound(pos.index + remaining));
@@ -311,16 +288,11 @@ fn minimize_attribute_changes(pos: &mut TextPosition, attrs: &TextAttributes) {
}
}
fn insert_item(
store: &mut DocStore,
ty: &mut YType,
pos: &mut TextPosition,
content: Content,
) -> JwstCodecResult {
if let Some(markers) = &ty.markers {
if content.countable() {
markers.update_marker_changes(pos.index, content.clock_len() as i64);
}
fn insert_item(store: &mut DocStore, ty: &mut YType, pos: &mut TextPosition, content: Content) -> JwstCodecResult {
if let Some(markers) = &ty.markers
&& content.countable()
{
markers.update_marker_changes(pos.index, content.clock_len() as i64);
}
let item = store.create_item(
@@ -383,14 +355,13 @@ fn insert_negated_attributes(
continue;
}
if let Content::Format { key, value } = &item.content {
if let Some(negated_value) = negated.get(key.as_str()) {
if negated_value == value {
negated.remove(key.as_str());
pos.forward();
continue;
}
}
if let Content::Format { key, value } = &item.content
&& let Some(negated_value) = negated.get(key.as_str())
&& negated_value == value
{
negated.remove(key.as_str());
pos.forward();
continue;
}
break;
@@ -488,12 +459,7 @@ fn format_text(
Ok(())
}
fn delete_text(
store: &mut DocStore,
ty: &mut YType,
pos: &mut TextPosition,
mut remaining: u64,
) -> JwstCodecResult {
fn delete_text(store: &mut DocStore, ty: &mut YType, pos: &mut TextPosition, mut remaining: u64) -> JwstCodecResult {
if remaining == 0 {
return Ok(());
}
@@ -501,19 +467,23 @@ fn delete_text(
let start = remaining;
while remaining > 0 {
let Some(item) = pos.right.get() else {
let item_ref = pos.right.clone();
let Some((indexable, item_len, item_id)) = item_ref.get().map(|item| (item.indexable(), item.len(), item.id))
else {
break;
};
if item.indexable() {
let item_len = item.len();
if indexable {
if remaining < item_len {
store.split_node(item.id, remaining)?;
store.split_node(item_id, remaining)?;
remaining = 0;
} else {
remaining -= item_len;
}
store.delete_item(item, Some(ty));
if let Some(item) = item_ref.get() {
store.delete_item(item, Some(ty));
}
}
pos.forward();
@@ -535,7 +505,7 @@ mod tests {
use super::{TextAttributes, TextDeltaOp, TextInsert};
#[cfg(not(loom))]
use crate::sync::{Arc, AtomicUsize, Ordering};
use crate::{loom_model, sync::thread, Any, Doc};
use crate::{Any, Doc, loom_model, sync::thread};
#[test]
fn test_manipulate_text() {
@@ -676,9 +646,7 @@ mod tests {
fn loom_parallel_ins_del_text() {
let seed = rand::rng().random();
let mut rand = ChaCha20Rng::seed_from_u64(seed);
let ranges = (0..20)
.map(|_| rand.random_range(0..16))
.collect::<Vec<_>>();
let ranges = (0..20).map(|_| rand.random_range(0..16)).collect::<Vec<_>>();
loom_model!({
let doc = Doc::new();
packages/common/y-octo/core/src/doc/types/value.rs
+2 -4
View File
@@ -2,7 +2,7 @@ use std::fmt::Display;
use super::*;
#[derive(Debug, PartialEq)]
#[derive(Debug, Clone, PartialEq)]
pub enum Value {
Any(Any),
Doc(Doc),
@@ -45,9 +45,7 @@ impl Value {
}
pub fn from_vec<T: Into<Any>>(el: Vec<T>) -> Self {
Value::Any(Any::Array(
el.into_iter().map(|item| item.into()).collect::<Vec<_>>(),
))
Value::Any(Any::Array(el.into_iter().map(|item| item.into()).collect::<Vec<_>>()))
}
}
packages/common/y-octo/core/src/doc/utils.rs
+19 -3
View File
@@ -16,9 +16,7 @@ pub fn encode_update_as_message(update: Vec<u8>) -> JwstCodecResult<Vec<u8>> {
Ok(buffer)
}
pub fn merge_updates_v1<V: AsRef<[u8]>, I: IntoIterator<Item = V>>(
updates: I,
) -> JwstCodecResult<Update> {
pub fn merge_updates_v1<V: AsRef<[u8]>, I: IntoIterator<Item = V>>(updates: I) -> JwstCodecResult<Update> {
let updates = updates
.into_iter()
.map(Update::decode_v1)
@@ -26,3 +24,21 @@ pub fn merge_updates_v1<V: AsRef<[u8]>, I: IntoIterator<Item = V>>(
Ok(Update::merge(updates))
}
/// It tends to generate small numbers.
/// Since the client id will be included in all crdt items, the
/// small client helps to reduce the binary size.
///
/// NOTE: The probability of 36% of the random number generated by
/// this function is greater than [u32::MAX]
pub fn prefer_small_random() -> u64 {
use rand::{distr::Distribution, rng};
use rand_distr::Exp;
let scale_factor = u16::MAX as f64;
let v: f64 = Exp::new(1.0 / scale_factor)
.map(|exp| exp.sample(&mut rng()))
.unwrap_or_else(|_| rand::random());
(v * scale_factor) as u64
}
packages/common/y-octo/core/src/lib.rs
+5 -7
View File
@@ -6,18 +6,16 @@ mod sync;
pub use codec::*;
pub use doc::{
encode_awareness_as_message, encode_update_as_message, merge_updates_v1, Any, Array, Awareness,
AwarenessEvent, Client, ClientMap, Clock, CrdtRead, CrdtReader, CrdtWrite, CrdtWriter, Doc,
DocOptions, HashMap as AHashMap, HashMapExt, History, HistoryOptions, Id, Map, RawDecoder,
RawEncoder, StateVector, StoreHistory, Text, TextAttributes, TextDelta, TextDeltaOp, TextInsert,
Update, Value,
Any, Array, Awareness, AwarenessEvent, Batch, Client, ClientMap, Clock, CrdtRead, CrdtReader, CrdtWrite, CrdtWriter,
Doc, DocOptions, HashMap as AHashMap, HashMapExt, History, HistoryOptions, Id, Map, RawDecoder, RawEncoder,
StateVector, StoreHistory, Text, TextAttributes, TextDelta, TextDeltaOp, TextInsert, Update, Value, batch_commit,
encode_awareness_as_message, encode_update_as_message, merge_updates_v1,
};
pub(crate) use doc::{Content, Item};
use log::{debug, warn};
use nom::IResult;
pub use protocol::{
read_sync_message, write_sync_message, AwarenessState, AwarenessStates, DocMessage, SyncMessage,
SyncMessageScanner,
AwarenessState, AwarenessStates, DocMessage, SyncMessage, SyncMessageScanner, read_sync_message, write_sync_message,
};
use thiserror::Error;
packages/common/y-octo/core/src/protocol/awareness.rs
+4 -14
View File
@@ -1,4 +1,4 @@
use nom::{multi::count, Parser};
use nom::{Parser, multi::count};
use super::*;
@@ -65,11 +65,7 @@ fn read_awareness_state(input: &[u8]) -> IResult<&[u8], (u64, AwarenessState)> {
Ok((tail, (client_id, AwarenessState { clock, content })))
}
fn write_awareness_state<W: Write>(
buffer: &mut W,
client_id: u64,
state: &AwarenessState,
) -> Result<(), IoError> {
fn write_awareness_state<W: Write>(buffer: &mut W, client_id: u64, state: &AwarenessState) -> Result<(), IoError> {
write_var_u64(buffer, client_id)?;
write_var_u64(buffer, state.clock)?;
write_var_string(buffer, state.content.clone())?;
@@ -118,14 +114,8 @@ mod tests {
];
let expected = HashMap::from([
(
1,
AwarenessState::new(5, String::from_utf8(vec![1]).unwrap()),
),
(
2,
AwarenessState::new(10, String::from_utf8(vec![2, 3]).unwrap()),
),
(1, AwarenessState::new(5, String::from_utf8(vec![1]).unwrap())),
(2, AwarenessState::new(10, String::from_utf8(vec![2, 3]).unwrap())),
(
5,
AwarenessState::new(5, String::from_utf8(vec![1, 2, 3, 4, 5]).unwrap()),
packages/common/y-octo/core/src/protocol/doc.rs
+3 -4
View File
@@ -5,10 +5,9 @@ use super::*;
#[cfg_attr(test, derive(proptest_derive::Arbitrary))]
pub enum DocMessage {
// state vector
// TODO: temporarily skipped in the test, because yrs decoding needs to ensure that the update
// in step1 is the correct state vector binary and any data can be included in our
// implementation (we will ensure the correctness of encoding and decoding in the subsequent
// decoding process)
// TODO: temporarily skipped in the test, because yrs decoding needs to ensure that the update in step1 is the
// correct state vector binary and any data can be included in our implementation (we will ensure the
// correctness of encoding and decoding in the subsequent decoding process)
#[cfg_attr(test, proptest(skip))]
Step1(Vec<u8>),
// update
packages/common/y-octo/core/src/protocol/mod.rs
+3 -3
View File
@@ -8,16 +8,16 @@ use std::{
io::{Error as IoError, Write},
};
use awareness::{read_awareness, write_awareness};
pub use awareness::{AwarenessState, AwarenessStates};
use awareness::{read_awareness, write_awareness};
pub use doc::DocMessage;
use doc::{read_doc_message, write_doc_message};
use log::debug;
use nom::{
error::{Error, ErrorKind},
IResult,
error::{Error, ErrorKind},
};
pub use scanner::SyncMessageScanner;
pub use sync::{read_sync_message, write_sync_message, SyncMessage};
pub use sync::{SyncMessage, read_sync_message, write_sync_message};
use super::*;
packages/common/y-octo/core/src/sync.rs
+4 -4
View File
@@ -1,21 +1,21 @@
pub use std::sync::{Arc, Weak};
#[allow(unused)]
#[cfg(not(loom))]
pub(crate) use std::sync::{
atomic::{AtomicBool, AtomicU16, AtomicU32, AtomicU8, Ordering},
Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard,
atomic::{AtomicBool, AtomicU8, AtomicU16, AtomicU32, Ordering},
};
pub use std::sync::{Arc, Weak};
#[cfg(all(test, not(loom)))]
pub(crate) use std::{
sync::{atomic::AtomicUsize, MutexGuard},
sync::{MutexGuard, atomic::AtomicUsize},
thread,
};
#[cfg(loom)]
pub(crate) use loom::{
sync::{
atomic::{AtomicBool, AtomicU16, AtomicU8, AtomicUsize, Ordering},
Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard,
atomic::{AtomicBool, AtomicU8, AtomicU16, AtomicU32, AtomicUsize, Ordering},
},
thread,
};