Struct Awareness

pub struct Awareness { /* private fields */ }

The Awareness class implements a simple shared state protocol that can be used for non-persistent data like awareness information (cursor, username, status, ..). Each client can update its own local state and listen to state changes of remote clients.

Each client is identified by a unique client id (something we borrow from doc.clientID). A client can override its own state by propagating a message with an increasing timestamp (clock). If such a message is received, it is applied if the known state of that client is older than the new state (clock < new_clock). If a client thinks that a remote client is offline, it may propagate a message with { clock, state: null, client }. If such a message is received, and the known clock of that client equals the received clock, it will clean the state.

Before a client disconnects, it should propagate a null state with an updated clock.

Implementations

impl Awareness

pub fn new(doc: Doc) -> Awareness

Creates a new instance of Awareness struct, which operates over a given document. Awareness instance has full ownership of that document. If necessary it can be accessed using either Awareness::doc or Awareness::doc_mut methods.

Examples found in repository

examples/client.rs (line 24)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 初始化日志
    tracing_subscriber::fmt().with_max_level(tracing::Level::INFO).init();

    // 从 `collaboration.rs` 测试中获取服务器详细信息
    let server_url = "ws://127.0.0.1:8080/collaboration"; // 确保端口与服务器测试匹配
    let room_name = "demo-room";

    // 1. 初始化客户端的文档和 awareness 状态
    let doc = Doc::new();
    let client_id = doc.client_id();
    let awareness: AwarenessRef =
        Arc::new(tokio::sync::RwLock::new(Awareness::new(doc)));

    // 🔍 生成唯一的用户 ID
    let unique_user_id =
        format!("client-user-{}-{}", chrono::Utc::now().timestamp(), client_id);

    tracing::info!("🆔 Yrs Client ID: {}", client_id);
    tracing::info!("👤 User ID: {}", unique_user_id);

    let mut provider = WebsocketProvider::new(
        server_url.to_string(),
        room_name.to_string(),
        awareness.clone(),
    )
    .await;

    // 订阅同步事件
    if let Some(mut receiver) = provider.subscribe_sync_events() {
        tokio::spawn(async move {
            while let Ok(event) = receiver.recv().await {
                match event {
                    SyncEvent::InitialSyncCompleted {
                        has_data,
                        elapsed_ms,
                    } => {
                        if has_data {
                            println!(
                                "🎉 同步完成，房间有数据！耗时: {}ms",
                                elapsed_ms
                            );
                        } else {
                            println!(
                                "📭 同步完成，空房间！耗时: {}ms",
                                elapsed_ms
                            );
                        }
                    },
                    SyncEvent::ProtocolStateChanged(state) => {
                        println!("📡 协议状态: {:?}", state);
                    },
                    SyncEvent::DataReceived => {
                        println!("📥 收到数据更新");
                    },
                    SyncEvent::ConnectionFailed(error) => {
                        println!("🔌 监听: {:?}", error);
                    },
                    SyncEvent::ConnectionChanged(status) => {
                        println!("🔌 连接状态: {:?}", status);
                    },
                    _ => {},
                }
            }
        });
    }

    // 3. 连接到服务器
    provider.connect().await;
    {
        let nodes_map = awareness.read().await.doc().get_or_insert_map("nodes");
        // 订阅 nodes变更 浅
        provider.subscription(nodes_map.observe(move |txn, event| {
            for (key, change) in event.keys(txn) {
                match change {
                    yrs::types::EntryChange::Inserted(value) => {
                        println!("新增 key: {}, value: {:?}", key, value);
                    },
                    yrs::types::EntryChange::Removed(old_value) => {
                        println!(
                            "删除 key: {}, old value: {:?}",
                            key, old_value
                        );
                    },
                    yrs::types::EntryChange::Updated(old_value, new_value) => {
                        println!(
                            "更新 key: {}, old: {:?}, new: {:?}",
                            key, old_value, new_value
                        );
                    },
                }
            }
        }));
        provider.subscription(nodes_map.observe_deep(move |txn, events| {
            for event in events.iter() {
                match event {
                    yrs::types::Event::Array(array_event) => {
                        // 更新了 标记数组 需要转换成 step
                    },
                    yrs::types::Event::Map(map_event) => {
                        // 更新了 节点属性 需要转换成 step 或者 添加节点
                    },
                    _ => {},
                }
            }
        }));
    }
    {
        let client_id_ref = client_id;
        let mut awareness_lock = awareness.write().await;
        provider.subscription(awareness_lock.on_update(move |event| {
            println!("📡 awareness update: {:?}", event.awareness_state());
            let states = event.awareness_state();
            for client_id in states.all_clients() {
                let meta: &yrs::sync::awareness::MetaClientState =
                    states.get_meta(client_id).unwrap();
                if client_id == client_id_ref {
                    // 本地客户端
                    println!(
                        "🏠 本地客户端 {}: clock={}, last_updated={:?}",
                        client_id, meta.clock, meta.last_updated
                    );
                } else {
                    // 远程客户端
                    println!(
                        "🌐 远程客户端 {}: clock={}, last_updated={:?}",
                        client_id, meta.clock, meta.last_updated
                    );
                }
            }
        }));

        // 🎯 使用唯一的用户 ID 避免状态覆盖
        awareness_lock.set_local_state(
            serde_json::json!({
                "user": {
                    "id": unique_user_id,
                    "name": "用户李兴栋",
                    "color": "#FFEAA7",
                    "online": true,
                    "client_id": client_id,
                    "timestamp": chrono::Utc::now().timestamp()
                },
                "online": true
            })
            .to_string(),
        );

        tracing::info!("✅ 设置 awareness 状态完成");
    }

    // 4. 事件循环
    let mut counter = 0; // 🔄 添加计数器确保状态变化

    loop {
        tokio::select! {

            // 定期添加本地更改以测试发送更新
            _ = time::sleep(Duration::from_secs(3)) => {
                if provider.is_connected() {
                    counter += 1; // 🔄 递增计数器

                    let mut awareness_lock = awareness.write().await;
                    awareness_lock.set_local_state(serde_json::json!({
                        "user": {
                            "id": unique_user_id,
                            "name": "用户李兴栋",
                            "color": "#FFEAA7",
                            "online": true,
                            "client_id": client_id,
                            "timestamp": chrono::Utc::now().timestamp(),
                            "heartbeat": counter  // 🔄 添加变化的字段
                        },
                        "online": true,
                        "last_activity": chrono::Utc::now().timestamp()  // 🔄 另一个变化字段
                    }).to_string());

                    let doc = awareness_lock.doc_mut();
                    let nodes_map = doc.get_or_insert_map("nodes");
                    let mut txn = doc.transact_mut_with(doc.client_id().to_string());
                    // 生成新节点 ID
            let node_id = uuid::Uuid::new_v4().to_string();

            // 简单地插入一个文本值作为节点内容
            let node_content = format!("{{\"type\": \"DXGC\", \"id\": \"{}\", \"client\": \"rust_client\"}}", node_id);
            nodes_map.insert(&mut txn, node_id.as_str(), node_content.as_str());

            // 事务会在 drop 时自动提交
            drop(txn);
                    tracing::info!("📝 已发送本地文档更改，heartbeat: {}", counter);
             }
            }
            // 处理 Ctrl-C 以优雅地断开连接
            _ = tokio::signal::ctrl_c() => {
                tracing::info!("🔌 正在断开连接...");
                provider.disconnect().await;
                tracing::info!("✅ 已断开连接。");
                break;
            }
        }
    }

    Ok(())
}

pub fn on_update<F>(&self, f: F) -> Subscription

where F: Fn(&Event) + 'static,

Returns a channel receiver for an incoming awareness events. This channel can be cloned.

Examples found in repository

examples/client.rs (lines 122-142)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 初始化日志
    tracing_subscriber::fmt().with_max_level(tracing::Level::INFO).init();

    // 从 `collaboration.rs` 测试中获取服务器详细信息
    let server_url = "ws://127.0.0.1:8080/collaboration"; // 确保端口与服务器测试匹配
    let room_name = "demo-room";

    // 1. 初始化客户端的文档和 awareness 状态
    let doc = Doc::new();
    let client_id = doc.client_id();
    let awareness: AwarenessRef =
        Arc::new(tokio::sync::RwLock::new(Awareness::new(doc)));

    // 🔍 生成唯一的用户 ID
    let unique_user_id =
        format!("client-user-{}-{}", chrono::Utc::now().timestamp(), client_id);

    tracing::info!("🆔 Yrs Client ID: {}", client_id);
    tracing::info!("👤 User ID: {}", unique_user_id);

    let mut provider = WebsocketProvider::new(
        server_url.to_string(),
        room_name.to_string(),
        awareness.clone(),
    )
    .await;

    // 订阅同步事件
    if let Some(mut receiver) = provider.subscribe_sync_events() {
        tokio::spawn(async move {
            while let Ok(event) = receiver.recv().await {
                match event {
                    SyncEvent::InitialSyncCompleted {
                        has_data,
                        elapsed_ms,
                    } => {
                        if has_data {
                            println!(
                                "🎉 同步完成，房间有数据！耗时: {}ms",
                                elapsed_ms
                            );
                        } else {
                            println!(
                                "📭 同步完成，空房间！耗时: {}ms",
                                elapsed_ms
                            );
                        }
                    },
                    SyncEvent::ProtocolStateChanged(state) => {
                        println!("📡 协议状态: {:?}", state);
                    },
                    SyncEvent::DataReceived => {
                        println!("📥 收到数据更新");
                    },
                    SyncEvent::ConnectionFailed(error) => {
                        println!("🔌 监听: {:?}", error);
                    },
                    SyncEvent::ConnectionChanged(status) => {
                        println!("🔌 连接状态: {:?}", status);
                    },
                    _ => {},
                }
            }
        });
    }

    // 3. 连接到服务器
    provider.connect().await;
    {
        let nodes_map = awareness.read().await.doc().get_or_insert_map("nodes");
        // 订阅 nodes变更 浅
        provider.subscription(nodes_map.observe(move |txn, event| {
            for (key, change) in event.keys(txn) {
                match change {
                    yrs::types::EntryChange::Inserted(value) => {
                        println!("新增 key: {}, value: {:?}", key, value);
                    },
                    yrs::types::EntryChange::Removed(old_value) => {
                        println!(
                            "删除 key: {}, old value: {:?}",
                            key, old_value
                        );
                    },
                    yrs::types::EntryChange::Updated(old_value, new_value) => {
                        println!(
                            "更新 key: {}, old: {:?}, new: {:?}",
                            key, old_value, new_value
                        );
                    },
                }
            }
        }));
        provider.subscription(nodes_map.observe_deep(move |txn, events| {
            for event in events.iter() {
                match event {
                    yrs::types::Event::Array(array_event) => {
                        // 更新了 标记数组 需要转换成 step
                    },
                    yrs::types::Event::Map(map_event) => {
                        // 更新了 节点属性 需要转换成 step 或者 添加节点
                    },
                    _ => {},
                }
            }
        }));
    }
    {
        let client_id_ref = client_id;
        let mut awareness_lock = awareness.write().await;
        provider.subscription(awareness_lock.on_update(move |event| {
            println!("📡 awareness update: {:?}", event.awareness_state());
            let states = event.awareness_state();
            for client_id in states.all_clients() {
                let meta: &yrs::sync::awareness::MetaClientState =
                    states.get_meta(client_id).unwrap();
                if client_id == client_id_ref {
                    // 本地客户端
                    println!(
                        "🏠 本地客户端 {}: clock={}, last_updated={:?}",
                        client_id, meta.clock, meta.last_updated
                    );
                } else {
                    // 远程客户端
                    println!(
                        "🌐 远程客户端 {}: clock={}, last_updated={:?}",
                        client_id, meta.clock, meta.last_updated
                    );
                }
            }
        }));

        // 🎯 使用唯一的用户 ID 避免状态覆盖
        awareness_lock.set_local_state(
            serde_json::json!({
                "user": {
                    "id": unique_user_id,
                    "name": "用户李兴栋",
                    "color": "#FFEAA7",
                    "online": true,
                    "client_id": client_id,
                    "timestamp": chrono::Utc::now().timestamp()
                },
                "online": true
            })
            .to_string(),
        );

        tracing::info!("✅ 设置 awareness 状态完成");
    }

    // 4. 事件循环
    let mut counter = 0; // 🔄 添加计数器确保状态变化

    loop {
        tokio::select! {

            // 定期添加本地更改以测试发送更新
            _ = time::sleep(Duration::from_secs(3)) => {
                if provider.is_connected() {
                    counter += 1; // 🔄 递增计数器

                    let mut awareness_lock = awareness.write().await;
                    awareness_lock.set_local_state(serde_json::json!({
                        "user": {
                            "id": unique_user_id,
                            "name": "用户李兴栋",
                            "color": "#FFEAA7",
                            "online": true,
                            "client_id": client_id,
                            "timestamp": chrono::Utc::now().timestamp(),
                            "heartbeat": counter  // 🔄 添加变化的字段
                        },
                        "online": true,
                        "last_activity": chrono::Utc::now().timestamp()  // 🔄 另一个变化字段
                    }).to_string());

                    let doc = awareness_lock.doc_mut();
                    let nodes_map = doc.get_or_insert_map("nodes");
                    let mut txn = doc.transact_mut_with(doc.client_id().to_string());
                    // 生成新节点 ID
            let node_id = uuid::Uuid::new_v4().to_string();

            // 简单地插入一个文本值作为节点内容
            let node_content = format!("{{\"type\": \"DXGC\", \"id\": \"{}\", \"client\": \"rust_client\"}}", node_id);
            nodes_map.insert(&mut txn, node_id.as_str(), node_content.as_str());

            // 事务会在 drop 时自动提交
            drop(txn);
                    tracing::info!("📝 已发送本地文档更改，heartbeat: {}", counter);
             }
            }
            // 处理 Ctrl-C 以优雅地断开连接
            _ = tokio::signal::ctrl_c() => {
                tracing::info!("🔌 正在断开连接...");
                provider.disconnect().await;
                tracing::info!("✅ 已断开连接。");
                break;
            }
        }
    }

    Ok(())
}

pub fn doc(&self) -> &Doc

Returns a read-only reference to an underlying Doc.

Examples found in repository

examples/client.rs (line 82)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 初始化日志
    tracing_subscriber::fmt().with_max_level(tracing::Level::INFO).init();

    // 从 `collaboration.rs` 测试中获取服务器详细信息
    let server_url = "ws://127.0.0.1:8080/collaboration"; // 确保端口与服务器测试匹配
    let room_name = "demo-room";

    // 1. 初始化客户端的文档和 awareness 状态
    let doc = Doc::new();
    let client_id = doc.client_id();
    let awareness: AwarenessRef =
        Arc::new(tokio::sync::RwLock::new(Awareness::new(doc)));

    // 🔍 生成唯一的用户 ID
    let unique_user_id =
        format!("client-user-{}-{}", chrono::Utc::now().timestamp(), client_id);

    tracing::info!("🆔 Yrs Client ID: {}", client_id);
    tracing::info!("👤 User ID: {}", unique_user_id);

    let mut provider = WebsocketProvider::new(
        server_url.to_string(),
        room_name.to_string(),
        awareness.clone(),
    )
    .await;

    // 订阅同步事件
    if let Some(mut receiver) = provider.subscribe_sync_events() {
        tokio::spawn(async move {
            while let Ok(event) = receiver.recv().await {
                match event {
                    SyncEvent::InitialSyncCompleted {
                        has_data,
                        elapsed_ms,
                    } => {
                        if has_data {
                            println!(
                                "🎉 同步完成，房间有数据！耗时: {}ms",
                                elapsed_ms
                            );
                        } else {
                            println!(
                                "📭 同步完成，空房间！耗时: {}ms",
                                elapsed_ms
                            );
                        }
                    },
                    SyncEvent::ProtocolStateChanged(state) => {
                        println!("📡 协议状态: {:?}", state);
                    },
                    SyncEvent::DataReceived => {
                        println!("📥 收到数据更新");
                    },
                    SyncEvent::ConnectionFailed(error) => {
                        println!("🔌 监听: {:?}", error);
                    },
                    SyncEvent::ConnectionChanged(status) => {
                        println!("🔌 连接状态: {:?}", status);
                    },
                    _ => {},
                }
            }
        });
    }

    // 3. 连接到服务器
    provider.connect().await;
    {
        let nodes_map = awareness.read().await.doc().get_or_insert_map("nodes");
        // 订阅 nodes变更 浅
        provider.subscription(nodes_map.observe(move |txn, event| {
            for (key, change) in event.keys(txn) {
                match change {
                    yrs::types::EntryChange::Inserted(value) => {
                        println!("新增 key: {}, value: {:?}", key, value);
                    },
                    yrs::types::EntryChange::Removed(old_value) => {
                        println!(
                            "删除 key: {}, old value: {:?}",
                            key, old_value
                        );
                    },
                    yrs::types::EntryChange::Updated(old_value, new_value) => {
                        println!(
                            "更新 key: {}, old: {:?}, new: {:?}",
                            key, old_value, new_value
                        );
                    },
                }
            }
        }));
        provider.subscription(nodes_map.observe_deep(move |txn, events| {
            for event in events.iter() {
                match event {
                    yrs::types::Event::Array(array_event) => {
                        // 更新了 标记数组 需要转换成 step
                    },
                    yrs::types::Event::Map(map_event) => {
                        // 更新了 节点属性 需要转换成 step 或者 添加节点
                    },
                    _ => {},
                }
            }
        }));
    }
    {
        let client_id_ref = client_id;
        let mut awareness_lock = awareness.write().await;
        provider.subscription(awareness_lock.on_update(move |event| {
            println!("📡 awareness update: {:?}", event.awareness_state());
            let states = event.awareness_state();
            for client_id in states.all_clients() {
                let meta: &yrs::sync::awareness::MetaClientState =
                    states.get_meta(client_id).unwrap();
                if client_id == client_id_ref {
                    // 本地客户端
                    println!(
                        "🏠 本地客户端 {}: clock={}, last_updated={:?}",
                        client_id, meta.clock, meta.last_updated
                    );
                } else {
                    // 远程客户端
                    println!(
                        "🌐 远程客户端 {}: clock={}, last_updated={:?}",
                        client_id, meta.clock, meta.last_updated
                    );
                }
            }
        }));

        // 🎯 使用唯一的用户 ID 避免状态覆盖
        awareness_lock.set_local_state(
            serde_json::json!({
                "user": {
                    "id": unique_user_id,
                    "name": "用户李兴栋",
                    "color": "#FFEAA7",
                    "online": true,
                    "client_id": client_id,
                    "timestamp": chrono::Utc::now().timestamp()
                },
                "online": true
            })
            .to_string(),
        );

        tracing::info!("✅ 设置 awareness 状态完成");
    }

    // 4. 事件循环
    let mut counter = 0; // 🔄 添加计数器确保状态变化

    loop {
        tokio::select! {

            // 定期添加本地更改以测试发送更新
            _ = time::sleep(Duration::from_secs(3)) => {
                if provider.is_connected() {
                    counter += 1; // 🔄 递增计数器

                    let mut awareness_lock = awareness.write().await;
                    awareness_lock.set_local_state(serde_json::json!({
                        "user": {
                            "id": unique_user_id,
                            "name": "用户李兴栋",
                            "color": "#FFEAA7",
                            "online": true,
                            "client_id": client_id,
                            "timestamp": chrono::Utc::now().timestamp(),
                            "heartbeat": counter  // 🔄 添加变化的字段
                        },
                        "online": true,
                        "last_activity": chrono::Utc::now().timestamp()  // 🔄 另一个变化字段
                    }).to_string());

                    let doc = awareness_lock.doc_mut();
                    let nodes_map = doc.get_or_insert_map("nodes");
                    let mut txn = doc.transact_mut_with(doc.client_id().to_string());
                    // 生成新节点 ID
            let node_id = uuid::Uuid::new_v4().to_string();

            // 简单地插入一个文本值作为节点内容
            let node_content = format!("{{\"type\": \"DXGC\", \"id\": \"{}\", \"client\": \"rust_client\"}}", node_id);
            nodes_map.insert(&mut txn, node_id.as_str(), node_content.as_str());

            // 事务会在 drop 时自动提交
            drop(txn);
                    tracing::info!("📝 已发送本地文档更改，heartbeat: {}", counter);
             }
            }
            // 处理 Ctrl-C 以优雅地断开连接
            _ = tokio::signal::ctrl_c() => {
                tracing::info!("🔌 正在断开连接...");
                provider.disconnect().await;
                tracing::info!("✅ 已断开连接。");
                break;
            }
        }
    }

    Ok(())
}

pub fn doc_mut(&mut self) -> &mut Doc

Returns a read-write reference to an underlying Doc.

Examples found in repository

examples/client.rs (line 189)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 初始化日志
    tracing_subscriber::fmt().with_max_level(tracing::Level::INFO).init();

    // 从 `collaboration.rs` 测试中获取服务器详细信息
    let server_url = "ws://127.0.0.1:8080/collaboration"; // 确保端口与服务器测试匹配
    let room_name = "demo-room";

    // 1. 初始化客户端的文档和 awareness 状态
    let doc = Doc::new();
    let client_id = doc.client_id();
    let awareness: AwarenessRef =
        Arc::new(tokio::sync::RwLock::new(Awareness::new(doc)));

    // 🔍 生成唯一的用户 ID
    let unique_user_id =
        format!("client-user-{}-{}", chrono::Utc::now().timestamp(), client_id);

    tracing::info!("🆔 Yrs Client ID: {}", client_id);
    tracing::info!("👤 User ID: {}", unique_user_id);

    let mut provider = WebsocketProvider::new(
        server_url.to_string(),
        room_name.to_string(),
        awareness.clone(),
    )
    .await;

    // 订阅同步事件
    if let Some(mut receiver) = provider.subscribe_sync_events() {
        tokio::spawn(async move {
            while let Ok(event) = receiver.recv().await {
                match event {
                    SyncEvent::InitialSyncCompleted {
                        has_data,
                        elapsed_ms,
                    } => {
                        if has_data {
                            println!(
                                "🎉 同步完成，房间有数据！耗时: {}ms",
                                elapsed_ms
                            );
                        } else {
                            println!(
                                "📭 同步完成，空房间！耗时: {}ms",
                                elapsed_ms
                            );
                        }
                    },
                    SyncEvent::ProtocolStateChanged(state) => {
                        println!("📡 协议状态: {:?}", state);
                    },
                    SyncEvent::DataReceived => {
                        println!("📥 收到数据更新");
                    },
                    SyncEvent::ConnectionFailed(error) => {
                        println!("🔌 监听: {:?}", error);
                    },
                    SyncEvent::ConnectionChanged(status) => {
                        println!("🔌 连接状态: {:?}", status);
                    },
                    _ => {},
                }
            }
        });
    }

    // 3. 连接到服务器
    provider.connect().await;
    {
        let nodes_map = awareness.read().await.doc().get_or_insert_map("nodes");
        // 订阅 nodes变更 浅
        provider.subscription(nodes_map.observe(move |txn, event| {
            for (key, change) in event.keys(txn) {
                match change {
                    yrs::types::EntryChange::Inserted(value) => {
                        println!("新增 key: {}, value: {:?}", key, value);
                    },
                    yrs::types::EntryChange::Removed(old_value) => {
                        println!(
                            "删除 key: {}, old value: {:?}",
                            key, old_value
                        );
                    },
                    yrs::types::EntryChange::Updated(old_value, new_value) => {
                        println!(
                            "更新 key: {}, old: {:?}, new: {:?}",
                            key, old_value, new_value
                        );
                    },
                }
            }
        }));
        provider.subscription(nodes_map.observe_deep(move |txn, events| {
            for event in events.iter() {
                match event {
                    yrs::types::Event::Array(array_event) => {
                        // 更新了 标记数组 需要转换成 step
                    },
                    yrs::types::Event::Map(map_event) => {
                        // 更新了 节点属性 需要转换成 step 或者 添加节点
                    },
                    _ => {},
                }
            }
        }));
    }
    {
        let client_id_ref = client_id;
        let mut awareness_lock = awareness.write().await;
        provider.subscription(awareness_lock.on_update(move |event| {
            println!("📡 awareness update: {:?}", event.awareness_state());
            let states = event.awareness_state();
            for client_id in states.all_clients() {
                let meta: &yrs::sync::awareness::MetaClientState =
                    states.get_meta(client_id).unwrap();
                if client_id == client_id_ref {
                    // 本地客户端
                    println!(
                        "🏠 本地客户端 {}: clock={}, last_updated={:?}",
                        client_id, meta.clock, meta.last_updated
                    );
                } else {
                    // 远程客户端
                    println!(
                        "🌐 远程客户端 {}: clock={}, last_updated={:?}",
                        client_id, meta.clock, meta.last_updated
                    );
                }
            }
        }));

        // 🎯 使用唯一的用户 ID 避免状态覆盖
        awareness_lock.set_local_state(
            serde_json::json!({
                "user": {
                    "id": unique_user_id,
                    "name": "用户李兴栋",
                    "color": "#FFEAA7",
                    "online": true,
                    "client_id": client_id,
                    "timestamp": chrono::Utc::now().timestamp()
                },
                "online": true
            })
            .to_string(),
        );

        tracing::info!("✅ 设置 awareness 状态完成");
    }

    // 4. 事件循环
    let mut counter = 0; // 🔄 添加计数器确保状态变化

    loop {
        tokio::select! {

            // 定期添加本地更改以测试发送更新
            _ = time::sleep(Duration::from_secs(3)) => {
                if provider.is_connected() {
                    counter += 1; // 🔄 递增计数器

                    let mut awareness_lock = awareness.write().await;
                    awareness_lock.set_local_state(serde_json::json!({
                        "user": {
                            "id": unique_user_id,
                            "name": "用户李兴栋",
                            "color": "#FFEAA7",
                            "online": true,
                            "client_id": client_id,
                            "timestamp": chrono::Utc::now().timestamp(),
                            "heartbeat": counter  // 🔄 添加变化的字段
                        },
                        "online": true,
                        "last_activity": chrono::Utc::now().timestamp()  // 🔄 另一个变化字段
                    }).to_string());

                    let doc = awareness_lock.doc_mut();
                    let nodes_map = doc.get_or_insert_map("nodes");
                    let mut txn = doc.transact_mut_with(doc.client_id().to_string());
                    // 生成新节点 ID
            let node_id = uuid::Uuid::new_v4().to_string();

            // 简单地插入一个文本值作为节点内容
            let node_content = format!("{{\"type\": \"DXGC\", \"id\": \"{}\", \"client\": \"rust_client\"}}", node_id);
            nodes_map.insert(&mut txn, node_id.as_str(), node_content.as_str());

            // 事务会在 drop 时自动提交
            drop(txn);
                    tracing::info!("📝 已发送本地文档更改，heartbeat: {}", counter);
             }
            }
            // 处理 Ctrl-C 以优雅地断开连接
            _ = tokio::signal::ctrl_c() => {
                tracing::info!("🔌 正在断开连接...");
                provider.disconnect().await;
                tracing::info!("✅ 已断开连接。");
                break;
            }
        }
    }

    Ok(())
}

pub fn client_id(&self) -> u64

Returns a globally unique client ID of an underlying Doc.

pub fn clients(&self) -> &HashMap<u64, String>

Returns a state map of all of the clients tracked by current Awareness instance. Those states are identified by their corresponding ClientIDs. The associated state is represented and replicated to other clients as a JSON string.

pub fn local_state(&self) -> Option<&str>

Returns a JSON string state representation of a current Awareness instance.

pub fn set_local_state<S>(&mut self, json: S)

where S: Into<String>,

Sets a current Awareness instance state to a corresponding JSON string. This state will be replicated to other clients as part of the AwarenessUpdate and it will trigger an event to be emitted if current instance was created using [Awareness::with_observer] method.

Examples found in repository

examples/client.rs (lines 145-158)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 初始化日志
    tracing_subscriber::fmt().with_max_level(tracing::Level::INFO).init();

    // 从 `collaboration.rs` 测试中获取服务器详细信息
    let server_url = "ws://127.0.0.1:8080/collaboration"; // 确保端口与服务器测试匹配
    let room_name = "demo-room";

    // 1. 初始化客户端的文档和 awareness 状态
    let doc = Doc::new();
    let client_id = doc.client_id();
    let awareness: AwarenessRef =
        Arc::new(tokio::sync::RwLock::new(Awareness::new(doc)));

    // 🔍 生成唯一的用户 ID
    let unique_user_id =
        format!("client-user-{}-{}", chrono::Utc::now().timestamp(), client_id);

    tracing::info!("🆔 Yrs Client ID: {}", client_id);
    tracing::info!("👤 User ID: {}", unique_user_id);

    let mut provider = WebsocketProvider::new(
        server_url.to_string(),
        room_name.to_string(),
        awareness.clone(),
    )
    .await;

    // 订阅同步事件
    if let Some(mut receiver) = provider.subscribe_sync_events() {
        tokio::spawn(async move {
            while let Ok(event) = receiver.recv().await {
                match event {
                    SyncEvent::InitialSyncCompleted {
                        has_data,
                        elapsed_ms,
                    } => {
                        if has_data {
                            println!(
                                "🎉 同步完成，房间有数据！耗时: {}ms",
                                elapsed_ms
                            );
                        } else {
                            println!(
                                "📭 同步完成，空房间！耗时: {}ms",
                                elapsed_ms
                            );
                        }
                    },
                    SyncEvent::ProtocolStateChanged(state) => {
                        println!("📡 协议状态: {:?}", state);
                    },
                    SyncEvent::DataReceived => {
                        println!("📥 收到数据更新");
                    },
                    SyncEvent::ConnectionFailed(error) => {
                        println!("🔌 监听: {:?}", error);
                    },
                    SyncEvent::ConnectionChanged(status) => {
                        println!("🔌 连接状态: {:?}", status);
                    },
                    _ => {},
                }
            }
        });
    }

    // 3. 连接到服务器
    provider.connect().await;
    {
        let nodes_map = awareness.read().await.doc().get_or_insert_map("nodes");
        // 订阅 nodes变更 浅
        provider.subscription(nodes_map.observe(move |txn, event| {
            for (key, change) in event.keys(txn) {
                match change {
                    yrs::types::EntryChange::Inserted(value) => {
                        println!("新增 key: {}, value: {:?}", key, value);
                    },
                    yrs::types::EntryChange::Removed(old_value) => {
                        println!(
                            "删除 key: {}, old value: {:?}",
                            key, old_value
                        );
                    },
                    yrs::types::EntryChange::Updated(old_value, new_value) => {
                        println!(
                            "更新 key: {}, old: {:?}, new: {:?}",
                            key, old_value, new_value
                        );
                    },
                }
            }
        }));
        provider.subscription(nodes_map.observe_deep(move |txn, events| {
            for event in events.iter() {
                match event {
                    yrs::types::Event::Array(array_event) => {
                        // 更新了 标记数组 需要转换成 step
                    },
                    yrs::types::Event::Map(map_event) => {
                        // 更新了 节点属性 需要转换成 step 或者 添加节点
                    },
                    _ => {},
                }
            }
        }));
    }
    {
        let client_id_ref = client_id;
        let mut awareness_lock = awareness.write().await;
        provider.subscription(awareness_lock.on_update(move |event| {
            println!("📡 awareness update: {:?}", event.awareness_state());
            let states = event.awareness_state();
            for client_id in states.all_clients() {
                let meta: &yrs::sync::awareness::MetaClientState =
                    states.get_meta(client_id).unwrap();
                if client_id == client_id_ref {
                    // 本地客户端
                    println!(
                        "🏠 本地客户端 {}: clock={}, last_updated={:?}",
                        client_id, meta.clock, meta.last_updated
                    );
                } else {
                    // 远程客户端
                    println!(
                        "🌐 远程客户端 {}: clock={}, last_updated={:?}",
                        client_id, meta.clock, meta.last_updated
                    );
                }
            }
        }));

        // 🎯 使用唯一的用户 ID 避免状态覆盖
        awareness_lock.set_local_state(
            serde_json::json!({
                "user": {
                    "id": unique_user_id,
                    "name": "用户李兴栋",
                    "color": "#FFEAA7",
                    "online": true,
                    "client_id": client_id,
                    "timestamp": chrono::Utc::now().timestamp()
                },
                "online": true
            })
            .to_string(),
        );

        tracing::info!("✅ 设置 awareness 状态完成");
    }

    // 4. 事件循环
    let mut counter = 0; // 🔄 添加计数器确保状态变化

    loop {
        tokio::select! {

            // 定期添加本地更改以测试发送更新
            _ = time::sleep(Duration::from_secs(3)) => {
                if provider.is_connected() {
                    counter += 1; // 🔄 递增计数器

                    let mut awareness_lock = awareness.write().await;
                    awareness_lock.set_local_state(serde_json::json!({
                        "user": {
                            "id": unique_user_id,
                            "name": "用户李兴栋",
                            "color": "#FFEAA7",
                            "online": true,
                            "client_id": client_id,
                            "timestamp": chrono::Utc::now().timestamp(),
                            "heartbeat": counter  // 🔄 添加变化的字段
                        },
                        "online": true,
                        "last_activity": chrono::Utc::now().timestamp()  // 🔄 另一个变化字段
                    }).to_string());

                    let doc = awareness_lock.doc_mut();
                    let nodes_map = doc.get_or_insert_map("nodes");
                    let mut txn = doc.transact_mut_with(doc.client_id().to_string());
                    // 生成新节点 ID
            let node_id = uuid::Uuid::new_v4().to_string();

            // 简单地插入一个文本值作为节点内容
            let node_content = format!("{{\"type\": \"DXGC\", \"id\": \"{}\", \"client\": \"rust_client\"}}", node_id);
            nodes_map.insert(&mut txn, node_id.as_str(), node_content.as_str());

            // 事务会在 drop 时自动提交
            drop(txn);
                    tracing::info!("📝 已发送本地文档更改，heartbeat: {}", counter);
             }
            }
            // 处理 Ctrl-C 以优雅地断开连接
            _ = tokio::signal::ctrl_c() => {
                tracing::info!("🔌 正在断开连接...");
                provider.disconnect().await;
                tracing::info!("✅ 已断开连接。");
                break;
            }
        }
    }

    Ok(())
}

pub fn remove_state(&mut self, client_id: u64)

Clears out a state of a given client, effectively marking it as disconnected.

pub fn clean_local_state(&mut self)

Clears out a state of a current client (see: Awareness::client_id), effectively marking it as disconnected.

pub fn update(&self) -> Result<AwarenessUpdate, Error>

Returns a serializable update object which is representation of a current Awareness state.

pub fn update_with_clients<I>( &self, clients: I, ) -> Result<AwarenessUpdate, Error>

where I: IntoIterator<Item = u64>,

Returns a serializable update object which is representation of a current Awareness state. Unlike Awareness::update, this method variant allows to prepare update only for a subset of known clients. These clients must all be known to a current Awareness instance, otherwise a Error::ClientNotFound error will be returned.

pub fn apply_update(&mut self, update: AwarenessUpdate) -> Result<(), Error>

Applies an update (incoming from remote channel or generated using Awareness::update / Awareness::update_with_clients methods) and modifies a state of a current instance.

If current instance has an observer channel (see: [Awareness::with_observer]), applied changes will also be emitted as events.

pub fn apply_update_summary( &mut self, update: AwarenessUpdate, ) -> Result<Option<AwarenessUpdateSummary>, Error>

Applies an update (incoming from remote channel or generated using Awareness::update / Awareness::update_with_clients methods) and modifies a state of a current instance. Returns an AwarenessUpdateSummary object informing about the changes that were applied.

If current instance has an observer channel (see: [Awareness::with_observer]), applied changes will also be emitted as events.

Trait Implementations

impl Debug for Awareness

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl Default for Awareness

fn default() -> Awareness

Returns the “default value” for a type.

impl Send for Awareness

impl Sync for Awareness