Rust实时数据同步库dittolive-ditto的使用指南

概述

dittolive-ditto是一个强大的实时数据同步库，专为跨平台设备间高效数据同步与协作而设计。它允许开发者在不同设备间实时同步数据，支持离线优先的工作模式，非常适合构建协作应用、多设备同步应用等场景。

主要特性

• 实时数据同步：设备间数据变更即时同步
• 离线优先：支持离线操作，网络恢复后自动同步
• 多平台支持：可在iOS、Android、macOS、Linux、Windows等平台使用
• 冲突解决：内置智能冲突解决机制
• 端到端加密：保障数据安全

安装方法

在Cargo.toml中添加依赖：

[dependencies]
ditto = "0.9"  # 请检查最新版本

基本使用方法

1. 初始化Ditto实例

use ditto::Ditto;

async fn init_ditto() -> Result<Ditto, Box<dyn std::error::Error>> {
    let ditto = Ditto::builder()
        .with_identity(|identity| {
            identity.online_with_authentication(|auth| {
                auth.app_id("YOUR_APP_ID".to_string())
                    .token("YOUR_AUTH_TOKEN".to_string())
            })
        })
        .build()?;
    
    ditto.start_sync().await?;
    Ok(ditto)
}

2. 创建和操作集合

use ditto::document::DocumentId;

async fn work_with_collection(ditto: &Ditto) {
    // 获取或创建集合
    let collection = ditto.store().collection("tasks").unwrap();
    
    // 插入文档
    let doc_id = DocumentId::new();
    collection
        .insert(
            doc_id,
            serde_json::json!({
                "title": "Buy milk",
                "completed": false,
                "created_at": chrono::Utc::now().to_rfc3339()
            }),
        )
        .await
        .unwrap();
    
    // 查询文档
    let tasks = collection.find("{}".to_string()).exec().await.unwrap();
    for task in tasks {
        println!("Task: {:?}", task);
    }
}

3. 监听数据变更

use ditto::subscription::Subscription;

async fn subscribe to_changes(ditto: &Ditto) {
    let collection = ditto.store().collection("tasks").unwrap();
    
    // 创建订阅
    let mut subscription = Subscription::new("live_tasks");
    subscription.set_query("{}".to_string());
    
    // 监听变更
    let mut live_query = collection
        .find("{}".to_string())
        .subscribe(&subscription)
        .await
        .unwrap();
    
    while let Some(event) = live_query.next().await {
        match event {
            ditto::live_query::LiveQueryEvent::Update(update) => {
                println!("Document updated: {:?}", update);
            }
            ditto::live_query::LiveQueryEvent::Insert(insert) => {
                println!("Document inserted: {:?}", insert);
            }
            ditto::live_query::LiveQueryEvent::Remove(remove) => {
                println!("Document removed: {:?}", remove);
            }
        }
    }
}

高级功能

1. 冲突解决

async fn handle_conflict(ditto: &Ditto) {
    let collection = ditto.store().collection("tasks").unwrap();
    
    // 设置冲突解决器
    collection
        .register_resolver("tasks", |conflict| {
            let local_doc = conflict.local();
            let remote_doc = conflict.remote();
            
            // 简单的基于时间戳的冲突解决策略
            let local_time = local_doc["updated_at"].as_str().unwrap();
            let remote_time = remote_doc["updated_at"].as_str().unwrap();
            
            if local_time > remote_time {
                conflict.resolve_with_local()
            } else {
                conflict.resolve_with_remote()
            }
        })
        .await;
}

2. 跨设备同步

async fn sync_across_devices(ditto: &Ditto) {
    // 连接到其他对等设备
    if let Err(e) = ditto.connect_peers().await {
        eprintln!("Failed to connect to peers: {}", e);
    }
    
    // 检查连接状态
    let peers = ditto.peers().await;
    println!("Connected peers: {:?}", peers);
}

完整示例代码

use ditto::{Ditto, DocumentId, live_query::LiveQueryEvent};
use serde_json::json;
use std::error::Error;
use tokio::signal;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    // 1. 初始化Ditto实例
    let ditto = Ditto::builder()
        .with_identity(|identity| {
            identity.online_with_authentication(|auth| {
                auth.app_id("my-collab-app".to_string())
                    .token("secure-token-123".to_string())
            })
        })
        .build()?;
    
    // 启动同步
    ditto.start_sync().await?;

    // 2. 创建并操作集合
    let collection = ditto.store().collection("collab_docs").unwrap();
    
    // 插入示例文档
    let doc_id = DocumentId::new();
    collection.insert(
        doc_id,
        json!({
            "content": "Initial document content",
            "version": 1,
            "last_updated": chrono::Utc::now().to_rfc3339()
        }),
    ).await?;

    // 3. 监听数据变更
    let mut live_query = collection
        .find("{}".to_string())
        .subscribe(&Subscription::new("doc_updates"))
        .await?;

    tokio::spawn(async move {
        while let Some(event) = live_query.next().await {
            match event {
                LiveQueryEvent::Update(update) => {
                    println!("文档更新: 版本 {}", update.document["version"]);
                }
                LiveQueryEvent::Insert(insert) => {
                    println!("新文档创建: ID {}", insert.document_id);
                }
                LiveQueryEvent::Remove(remove) => {
                    println!("文档删除: ID {}", remove.document_id);
                }
            }
        }
    });

    // 4. 高级功能示例 - 冲突解决
    collection.register_resolver("collab_docs", |conflict| {
        let local_ver = conflict.local()["version"].as_u64().unwrap_or(0);
        let remote_ver = conflict.remote()["version"].as_u64().unwrap_or(0);
        
        if local_ver > remote_ver {
            conflict.resolve_with_local()
        } else {
            conflict.resolve_with_remote()
        }
    }).await;

    // 5. 跨设备同步
    if let Err(e) = ditto.connect_peers().await {
        eprintln!("Peer连接错误: {}", e);
    }

    // 保持应用运行直到收到Ctrl-C
    signal::ctrl_c().await?;
    println!("正在关闭应用...");
    Ok(())
}

注意事项

1. 认证信息：生产环境中不要硬编码认证信息，应从安全配置中获取
2. 错误处理：实际应用中应妥善处理所有可能的错误
3. 性能考虑：大数据集时应使用适当的查询条件和索引
4. 网络状态：考虑处理网络连接断开和恢复的情况

dittolive-ditto为Rust开发者提供了强大的实时数据同步能力，可以大大简化跨设备协作应用的开发工作。