dittolive-ditto 5.0.3

#![allow(clippy::disallowed_macros)]

use std::{str::FromStr, sync::Arc};

use anyhow::Result;
use dittolive_ditto::{
    fs::TempRoot,
    prelude::*,
    preview::{datastreams::WriteOnly, peer_pubkey::PeerPubkey},
};

#[tokio::main]
async fn main() -> Result<()> {
    let root = TempRoot::new();
    let ditto = Ditto::open_sync(
        DittoConfig::new(
            "bus_example_app",
            DittoConfigConnect::SmallPeersOnly { private_key: None },
        )
        .with_persistence_directory(root.root_path()),
    )?;
    ditto.set_license_from_env("DITTO_LICENSE")?;

    DittoLogger::set_minimum_log_level(LogLevel::Debug);
    ditto.update_transport_config(|tc| {
        tc.enable_all_peer_to_peer();
    });

    ditto.sync().start()?;
    let ditto = Arc::new(ditto);

    let _acceptor = ditto
        .datastreams()
        .bind_topic("example")
        .on_receive_factory(tokio::sync::mpsc::unbounded_channel)
        .finish_with(|mut stream| {
            tokio::task::spawn(async move {
                let pk = stream.peer_pubkey();
                while let Some(payload) = stream.recv().await {
                    let msg = String::from_utf8_lossy(&payload);

                    match &*msg {
                        "Hello, world!" => {
                            println!("Received hello from peer: {}", pk);
                            stream.message("Hello back!").send();
                        }
                        _ => {
                            println!("Received message: {}", msg);
                        }
                    }
                }
            });
        });

    // wait for another peer to join and grab its PeerPubkey
    let remote_peer = loop {
        let peers = ditto.presence().graph().remote_peers;

        if let Some(peer) = peers.iter().next() {
            break PeerPubkey::from_str(&peer.peer_key).unwrap();
        }

        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    };

    let stream = ditto
        .datastreams()
        .connect(remote_peer.clone(), "example")
        .on_receive_factory(WriteOnly)
        .finish_async()
        .await
        .unwrap();
    loop {
        println!("Sending hello to: {}", &remote_peer);

        stream.message("Hello, world!").send();

        tokio::time::sleep(std::time::Duration::from_secs(10)).await;
    }
}