quick-start-rust-iroh

main-sync.rs

use std::collections::HashSet;

use anyhow::Result;
use futures_lite::stream::StreamExt;
use iroh::{
    base::node_addr::AddrInfoOptions,
    client::{
        docs::{LiveEvent, ShareMode},
        MemDoc,
    },
    docs::{store::Query, DocTicket},
    node::MemNode,
};

#[tokio::main]
async fn main() -> Result<()> {
    let node = iroh::node::Node::memory().spawn().await?;

    let node_id = node.node_id();
    println!("Started iroh node with id:\n{}", node_id);

    let maybe_ticket = std::env::args().nth(1);
    if let Some(ticket) = maybe_ticket {
        join_document(node, ticket).await?;
    } else {
        create_document(node).await?;
        tokio::signal::ctrl_c().await?;
    }

    Ok(())
}

async fn create_document(node: MemNode) -> Result<()> {
    let doc = node.docs.create().await?;
    println!("Created doc {}", doc.id());

    let author = node.authors.create().await?;
    println!("Created author: {}", author);

    doc.set_bytes(author, "hello", "world").await?;

    let maybe_entry = doc.get_one(Query::key_exact("hello")).await?;
    // unwrap, because we know we have this entry
    let entry = maybe_entry.unwrap();

    let key = std::str::from_utf8(&entry.key()).unwrap();

    let value_bytes = entry.content_bytes(&*node).await?;
    let value = std::str::from_utf8(&value_bytes).unwrap();

    println!("Stored and retrieved entry: {} => {}", key, value);

    let ticket = doc.share(ShareMode::Write, AddrInfoOptions::Id).await?;

    println!("Doc ticket:\n{}", ticket);
    Ok(())
}

async fn join_document(node: MemNode, ticket: String) -> Result<()> {
    let ticket: DocTicket = ticket.parse()?;

    let doc = node.docs.import(ticket).await?;
    wait_for_sync(&doc).await?;

    let mut entries = doc.get_many(Query::all()).await?;
    while let Some(entry) = entries.next().await {
        let entry = entry?;
        let key = std::str::from_utf8(&entry.key()).unwrap();
        let value_bytes = entry.content_bytes(&*node).await?;
        let value = std::str::from_utf8(&value_bytes).unwrap();

        println!("{} => {}", key, value);
    }

    Ok(())
}

/// Wait for a sync to finish, and all blobs to be received
///
/// This is racy and should be part of the iroh API.
/// - Racy because sync could have finished before this starts.
/// - should be part of the iroh API because it requires a lot of knowledge
///   to write correctly
async fn wait_for_sync(doc: &MemDoc) -> Result<()> {
    let mut seen_blobs = HashSet::new();

    let mut events = doc.subscribe().await?;
    while let Some(event) = events.next().await {
        let event = event?;
        if matches!(event, LiveEvent::SyncFinished(_)) {
            println!("Sync finished");
            break;
        }
        if let LiveEvent::ContentReady { hash } = event {
            seen_blobs.insert(hash);
        }
    }

    let mut blobs: HashSet<_> = doc
        .get_many(Query::all())
        .await?
        .map(|e| e.map(|e| e.content_hash()))
        .try_collect()
        .await?;

    for blob in &seen_blobs {
        blobs.remove(blob);
    }

    while let Some(event) = events.next().await {
        let event = event?;
        if let LiveEvent::ContentReady { hash } = event {
            blobs.remove(&hash);
            if blobs.is_empty() {
                // we have the content for all blobs
                break;
            }
        }
    }

    Ok(())
}

main.rs

use anyhow::Result;
use futures_lite::stream::StreamExt;
use iroh::{
    base::node_addr::AddrInfoOptions,
    client::docs::{LiveEvent, ShareMode},
    docs::{store::Query, DocTicket},
    node::MemNode,
};

#[tokio::main]
async fn main() -> Result<()> {
    let node = iroh::node::Node::memory().spawn().await?;

    let node_id = node.node_id();
    println!("Started iroh node with id:\n{}", node_id);

    let maybe_ticket = std::env::args().nth(1);
    if let Some(ticket) = maybe_ticket {
        join_document(node, ticket).await?;
    } else {
        create_document(node).await?;
        tokio::signal::ctrl_c().await?;
    }

    Ok(())
}

async fn create_document(node: MemNode) -> Result<()> {
    let doc = node.docs.create().await?;
    println!("Created doc {}", doc.id());

    let author = node.authors.create().await?;
    println!("Created author: {}", author);

    doc.set_bytes(author, "hello", "world").await?;

    let maybe_entry = doc.get_one(Query::key_exact("hello")).await?;
    // unwrap, because we know we have this entry
    let entry = maybe_entry.unwrap();

    let key = std::str::from_utf8(&entry.key()).unwrap();

    let value_bytes = entry.content_bytes(&*node).await?;
    let value = std::str::from_utf8(&value_bytes).unwrap();

    println!("Stored and retrieved entry: {} => {}", key, value);

    let ticket = doc.share(ShareMode::Write, AddrInfoOptions::Id).await?;

    println!("Doc ticket:\n{}", ticket);
    Ok(())
}

async fn join_document(node: MemNode, ticket: String) -> Result<()> {
    let ticket: DocTicket = ticket.parse()?;

    let doc = node.docs.import(ticket).await?;

    // This is racy and shitty API
    // what we need is sth like
    // doc.wait_for_sync_and_blobs_done().await?;
    // especially on first import
    let mut events = doc.subscribe().await?;
    while let Some(event) = events.next().await {
        let event = event?;
        if matches!(event, LiveEvent::SyncFinished(_)) {
            println!("Sync finished");
            break;
        }
    }

    let mut entries = doc.get_many(Query::all()).await?;

    while let Some(entry) = entries.next().await {
        let entry = entry?;
        let key = std::str::from_utf8(&entry.key()).unwrap();

        // No blob :sadface:
        println!("{}", key);
    }

    Ok(())
}