Struct ProxyHandle 

pub struct ProxyHandle { /* private fields */ }

Implementations

impl ProxyHandle

pub fn runtime(&self) -> Result<Arc<ProxyRuntime>, KoiError>

Examples found in repository

examples/embedded-integration.rs (line 856)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args: Vec<String> = std::env::args().collect();
    if has_flag(&args, "--help") {
        println!(
            "Usage: cargo run -p koi-embedded --example embedded-integration -- [options]\n\nOptions:\n  --timeout N   Overall timeout in seconds (default: 30)\n  --verbose     Verbose logging"
        );
        return Ok(());
    }

    let verbose = has_flag(&args, "--verbose");
    let with_certmesh = true;
    let skip_mdns = false;
    let timeout_secs = read_arg_value(&args, "--timeout")
        .and_then(|value| value.parse::<u64>().ok())
        .unwrap_or(30);
    let total_deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);

    let data_dir = temp_data_dir();
    let mut harness = Harness::new(verbose);

    harness.log(format!("data dir: {}", data_dir.display()));
    harness.log(format!("mdns: {}", !skip_mdns));
    harness.log(format!("certmesh: {}", with_certmesh));

    let koi = Builder::new()
        .data_dir(&data_dir)
        .service_mode(ServiceMode::EmbeddedOnly)
        .mdns(!skip_mdns)
        .dns_enabled(true)
        .dns(|cfg| cfg.port(15353))
        .health(true)
        .certmesh(with_certmesh)
        .proxy(true)
        .build()?;
    let handle = koi.start().await?;

    let mdns = match handle.mdns() {
        Ok(mdns) => mdns,
        Err(err) => {
            harness.fail("mdns: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let dns = match handle.dns() {
        Ok(dns) => dns,
        Err(err) => {
            harness.fail("dns: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let health = match handle.health() {
        Ok(health) => health,
        Err(err) => {
            harness.fail("health: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let proxy = match handle.proxy() {
        Ok(proxy) => proxy,
        Err(err) => {
            harness.fail("proxy: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let certmesh = match handle.certmesh() {
        Ok(certmesh) => certmesh,
        Err(err) => {
            harness.fail("certmesh: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };

    let (http_addr, http_cancel) = start_http_server(
        mdns.core()?,
        dns.runtime()?,
        health.core()?,
        certmesh.core()?,
        proxy.runtime()?,
    )
    .await?;
    let http_base = format!("http://{}", http_addr);
    harness.log(format!("http base: {http_base}"));

    // DNS: add entry, lookup, and event.
    let mut rx = handle.subscribe();
    let entry = DnsEntry {
        name: "embedded-test.lan".to_string(),
        ip: "127.0.0.1".to_string(),
        ttl: None,
    };
    let _ = dns.add_entry(entry)?;
    let event = wait_for_event(
        &mut rx,
        Duration::from_secs(2),
        |event| matches!(event, KoiEvent::DnsEntryUpdated { name, .. } if name == "embedded-test.lan"),
    )
    .await;
    if event.is_some() {
        harness.pass("dns: event emitted");
    } else {
        harness.fail("dns: event emitted", "no DnsEntryUpdated event received");
    }

    let result = dns
        .lookup("embedded-test.lan", hickory_proto::rr::RecordType::A)
        .await;
    match result {
        Some(result) => {
            if result.ips.contains(&IpAddr::from([127, 0, 0, 1])) && result.source == "static" {
                harness.pass("dns: lookup static entry");
            } else {
                harness.fail("dns: lookup static entry", "unexpected lookup result");
            }
        }
        None => harness.fail("dns: lookup static entry", "lookup returned none"),
    }

    let names = dns.list_names();
    if names.iter().any(|name| name == "embedded-test.lan.") {
        harness.pass("dns: list names includes entry");
    } else {
        harness.fail("dns: list names includes entry", "name missing from list");
    }

    let _ = dns.remove_entry("embedded-test.lan");
    let removed_event = wait_for_event(
        &mut rx,
        Duration::from_secs(2),
        |event| matches!(event, KoiEvent::DnsEntryRemoved { name } if name == "embedded-test.lan"),
    )
    .await;
    if removed_event.is_some() {
        harness.pass("dns: remove emits removal event");
    } else {
        harness.fail(
            "dns: remove emits removal event",
            "no removal event received",
        );
    }

    // Health: run a TCP check against a local listener.
    let mut rx = handle.subscribe();
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    tokio::spawn(async move {
        loop {
            let _ = listener.accept().await;
        }
    });

    let check = HealthCheck {
        name: "tcp-local".to_string(),
        kind: ServiceCheckKind::Tcp,
        target: format!("127.0.0.1:{}", addr.port()),
        interval_secs: 1,
        timeout_secs: 1,
    };
    health.add_check(check).await?;
    health.core()?.run_checks_once().await;
    let snapshot = health.status().await;
    let status = snapshot
        .services
        .iter()
        .find(|svc| svc.name == "tcp-local")
        .map(|svc| svc.status);
    match status {
        Some(ServiceStatus::Up) => harness.pass("health: tcp check up"),
        Some(other) => harness.fail(
            "health: tcp check up",
            &format!("unexpected status: {other:?}"),
        ),
        None => harness.fail("health: tcp check up", "service missing"),
    }

    let event = wait_for_event(
        &mut rx,
        Duration::from_secs(3),
        |event| matches!(event, KoiEvent::HealthChanged { name, .. } if name == "tcp-local"),
    )
    .await;
    if event.is_some() {
        harness.pass("health: event emitted");
    } else {
        harness.fail("health: event emitted", "no HealthChanged event received");
    }

    let _ = health.remove_check("tcp-local").await;
    let snapshot = health.status().await;
    if snapshot.services.iter().any(|svc| svc.name == "tcp-local") {
        harness.fail("health: remove check", "check still present after removal");
    } else {
        harness.pass("health: remove check");
    }

    // mDNS: register + browse.
    let browse = mdns.browse("_koi._tcp").await;
    if let Ok(browse) = browse {
        let mut txt = HashMap::new();
        txt.insert("source".to_string(), "embedded".to_string());
        let payload = RegisterPayload {
            name: "koi-embedded-test".to_string(),
            service_type: "_koi._tcp".to_string(),
            port: 51515,
            ip: Some("127.0.0.1".to_string()),
            lease_secs: Some(30),
            txt,
        };
        let reg = mdns.register(payload);
        if let Ok(reg) = reg {
            let found = tokio::time::timeout(Duration::from_secs(5), browse.recv()).await;
            match found {
                Ok(Some(_event)) => harness.pass("mdns: register + browse"),
                Ok(None) => harness.fail("mdns: register + browse", "browse stream ended"),
                Err(_) => harness.fail("mdns: register + browse", "no events within timeout"),
            }

            match mdns.resolve("koi-embedded-test._koi._tcp.local.").await {
                Ok(record) if record.port == Some(51515) => {
                    harness.pass("mdns: resolve registered service");
                }
                Ok(_) => harness.fail("mdns: resolve registered service", "unexpected record"),
                Err(err) => harness.fail("mdns: resolve registered service", &format!("{err}")),
            }

            match mdns.unregister(&reg.id) {
                Ok(()) => harness.pass("mdns: unregister"),
                Err(err) => harness.fail("mdns: unregister", &format!("{err}")),
            }
        } else {
            harness.fail("mdns: register + browse", "register failed");
        }
    } else {
        harness.fail("mdns: register + browse", "browse failed");
    }

    // Proxy: upsert and read entries.
    let mut rx = handle.subscribe();
    let entry = ProxyEntry {
        name: "embedded-proxy".to_string(),
        listen_port: 18080,
        backend: "http://127.0.0.1:18081".to_string(),
        allow_remote: false,
    };
    let result = proxy.upsert(entry.clone()).await;
    if result.is_ok() {
        let entries = proxy.entries().await;
        if entries.iter().any(|item| item.name == entry.name) {
            harness.pass("proxy: upsert entry");
        } else {
            harness.fail("proxy: upsert entry", "entry missing after upsert");
        }
        let event = wait_for_event(&mut rx, Duration::from_secs(3), |event| {
            matches!(event, KoiEvent::ProxyEntryUpdated { entry } if entry.name == "embedded-proxy")
        })
        .await;
        if event.is_some() {
            harness.pass("proxy: event emitted");
        } else {
            harness.fail(
                "proxy: event emitted",
                "no ProxyEntryUpdated event received",
            );
        }
        let _ = proxy.remove("embedded-proxy").await;
        let entries = proxy.entries().await;
        if entries.iter().any(|item| item.name == "embedded-proxy") {
            harness.fail("proxy: remove entry", "entry still present after removal");
        } else {
            harness.pass("proxy: remove entry");
        }
    } else {
        harness.fail("proxy: upsert entry", "upsert failed");
    }

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(10))
        .build()?;
    if let Err(err) = run_http_tests(&http_base, &client, &mut harness).await {
        harness.fail("http: suite", &format!("{err}"));
    }

    if let Err(err) = run_ipc_tests(mdns.core()?, &mut harness).await {
        harness.fail("ipc: suite", &format!("{err}"));
    }

    http_cancel.cancel();
    handle.shutdown().await?;
    if tokio::time::Instant::now() > total_deadline {
        harness.fail("runtime", "overall timeout exceeded");
    }
    harness.summary();
    if harness.failed > 0 {
        std::process::exit(1);
    }
    Ok(())
}

pub fn core(&self) -> Result<Arc<ProxyCore>, KoiError>

pub async fn entries(&self) -> Vec<ProxyEntry>

Examples found in repository

examples/embedded-integration.rs (line 1024)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args: Vec<String> = std::env::args().collect();
    if has_flag(&args, "--help") {
        println!(
            "Usage: cargo run -p koi-embedded --example embedded-integration -- [options]\n\nOptions:\n  --timeout N   Overall timeout in seconds (default: 30)\n  --verbose     Verbose logging"
        );
        return Ok(());
    }

    let verbose = has_flag(&args, "--verbose");
    let with_certmesh = true;
    let skip_mdns = false;
    let timeout_secs = read_arg_value(&args, "--timeout")
        .and_then(|value| value.parse::<u64>().ok())
        .unwrap_or(30);
    let total_deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);

    let data_dir = temp_data_dir();
    let mut harness = Harness::new(verbose);

    harness.log(format!("data dir: {}", data_dir.display()));
    harness.log(format!("mdns: {}", !skip_mdns));
    harness.log(format!("certmesh: {}", with_certmesh));

    let koi = Builder::new()
        .data_dir(&data_dir)
        .service_mode(ServiceMode::EmbeddedOnly)
        .mdns(!skip_mdns)
        .dns_enabled(true)
        .dns(|cfg| cfg.port(15353))
        .health(true)
        .certmesh(with_certmesh)
        .proxy(true)
        .build()?;
    let handle = koi.start().await?;

    let mdns = match handle.mdns() {
        Ok(mdns) => mdns,
        Err(err) => {
            harness.fail("mdns: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let dns = match handle.dns() {
        Ok(dns) => dns,
        Err(err) => {
            harness.fail("dns: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let health = match handle.health() {
        Ok(health) => health,
        Err(err) => {
            harness.fail("health: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let proxy = match handle.proxy() {
        Ok(proxy) => proxy,
        Err(err) => {
            harness.fail("proxy: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let certmesh = match handle.certmesh() {
        Ok(certmesh) => certmesh,
        Err(err) => {
            harness.fail("certmesh: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };

    let (http_addr, http_cancel) = start_http_server(
        mdns.core()?,
        dns.runtime()?,
        health.core()?,
        certmesh.core()?,
        proxy.runtime()?,
    )
    .await?;
    let http_base = format!("http://{}", http_addr);
    harness.log(format!("http base: {http_base}"));

    // DNS: add entry, lookup, and event.
    let mut rx = handle.subscribe();
    let entry = DnsEntry {
        name: "embedded-test.lan".to_string(),
        ip: "127.0.0.1".to_string(),
        ttl: None,
    };
    let _ = dns.add_entry(entry)?;
    let event = wait_for_event(
        &mut rx,
        Duration::from_secs(2),
        |event| matches!(event, KoiEvent::DnsEntryUpdated { name, .. } if name == "embedded-test.lan"),
    )
    .await;
    if event.is_some() {
        harness.pass("dns: event emitted");
    } else {
        harness.fail("dns: event emitted", "no DnsEntryUpdated event received");
    }

    let result = dns
        .lookup("embedded-test.lan", hickory_proto::rr::RecordType::A)
        .await;
    match result {
        Some(result) => {
            if result.ips.contains(&IpAddr::from([127, 0, 0, 1])) && result.source == "static" {
                harness.pass("dns: lookup static entry");
            } else {
                harness.fail("dns: lookup static entry", "unexpected lookup result");
            }
        }
        None => harness.fail("dns: lookup static entry", "lookup returned none"),
    }

    let names = dns.list_names();
    if names.iter().any(|name| name == "embedded-test.lan.") {
        harness.pass("dns: list names includes entry");
    } else {
        harness.fail("dns: list names includes entry", "name missing from list");
    }

    let _ = dns.remove_entry("embedded-test.lan");
    let removed_event = wait_for_event(
        &mut rx,
        Duration::from_secs(2),
        |event| matches!(event, KoiEvent::DnsEntryRemoved { name } if name == "embedded-test.lan"),
    )
    .await;
    if removed_event.is_some() {
        harness.pass("dns: remove emits removal event");
    } else {
        harness.fail(
            "dns: remove emits removal event",
            "no removal event received",
        );
    }

    // Health: run a TCP check against a local listener.
    let mut rx = handle.subscribe();
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    tokio::spawn(async move {
        loop {
            let _ = listener.accept().await;
        }
    });

    let check = HealthCheck {
        name: "tcp-local".to_string(),
        kind: ServiceCheckKind::Tcp,
        target: format!("127.0.0.1:{}", addr.port()),
        interval_secs: 1,
        timeout_secs: 1,
    };
    health.add_check(check).await?;
    health.core()?.run_checks_once().await;
    let snapshot = health.status().await;
    let status = snapshot
        .services
        .iter()
        .find(|svc| svc.name == "tcp-local")
        .map(|svc| svc.status);
    match status {
        Some(ServiceStatus::Up) => harness.pass("health: tcp check up"),
        Some(other) => harness.fail(
            "health: tcp check up",
            &format!("unexpected status: {other:?}"),
        ),
        None => harness.fail("health: tcp check up", "service missing"),
    }

    let event = wait_for_event(
        &mut rx,
        Duration::from_secs(3),
        |event| matches!(event, KoiEvent::HealthChanged { name, .. } if name == "tcp-local"),
    )
    .await;
    if event.is_some() {
        harness.pass("health: event emitted");
    } else {
        harness.fail("health: event emitted", "no HealthChanged event received");
    }

    let _ = health.remove_check("tcp-local").await;
    let snapshot = health.status().await;
    if snapshot.services.iter().any(|svc| svc.name == "tcp-local") {
        harness.fail("health: remove check", "check still present after removal");
    } else {
        harness.pass("health: remove check");
    }

    // mDNS: register + browse.
    let browse = mdns.browse("_koi._tcp").await;
    if let Ok(browse) = browse {
        let mut txt = HashMap::new();
        txt.insert("source".to_string(), "embedded".to_string());
        let payload = RegisterPayload {
            name: "koi-embedded-test".to_string(),
            service_type: "_koi._tcp".to_string(),
            port: 51515,
            ip: Some("127.0.0.1".to_string()),
            lease_secs: Some(30),
            txt,
        };
        let reg = mdns.register(payload);
        if let Ok(reg) = reg {
            let found = tokio::time::timeout(Duration::from_secs(5), browse.recv()).await;
            match found {
                Ok(Some(_event)) => harness.pass("mdns: register + browse"),
                Ok(None) => harness.fail("mdns: register + browse", "browse stream ended"),
                Err(_) => harness.fail("mdns: register + browse", "no events within timeout"),
            }

            match mdns.resolve("koi-embedded-test._koi._tcp.local.").await {
                Ok(record) if record.port == Some(51515) => {
                    harness.pass("mdns: resolve registered service");
                }
                Ok(_) => harness.fail("mdns: resolve registered service", "unexpected record"),
                Err(err) => harness.fail("mdns: resolve registered service", &format!("{err}")),
            }

            match mdns.unregister(&reg.id) {
                Ok(()) => harness.pass("mdns: unregister"),
                Err(err) => harness.fail("mdns: unregister", &format!("{err}")),
            }
        } else {
            harness.fail("mdns: register + browse", "register failed");
        }
    } else {
        harness.fail("mdns: register + browse", "browse failed");
    }

    // Proxy: upsert and read entries.
    let mut rx = handle.subscribe();
    let entry = ProxyEntry {
        name: "embedded-proxy".to_string(),
        listen_port: 18080,
        backend: "http://127.0.0.1:18081".to_string(),
        allow_remote: false,
    };
    let result = proxy.upsert(entry.clone()).await;
    if result.is_ok() {
        let entries = proxy.entries().await;
        if entries.iter().any(|item| item.name == entry.name) {
            harness.pass("proxy: upsert entry");
        } else {
            harness.fail("proxy: upsert entry", "entry missing after upsert");
        }
        let event = wait_for_event(&mut rx, Duration::from_secs(3), |event| {
            matches!(event, KoiEvent::ProxyEntryUpdated { entry } if entry.name == "embedded-proxy")
        })
        .await;
        if event.is_some() {
            harness.pass("proxy: event emitted");
        } else {
            harness.fail(
                "proxy: event emitted",
                "no ProxyEntryUpdated event received",
            );
        }
        let _ = proxy.remove("embedded-proxy").await;
        let entries = proxy.entries().await;
        if entries.iter().any(|item| item.name == "embedded-proxy") {
            harness.fail("proxy: remove entry", "entry still present after removal");
        } else {
            harness.pass("proxy: remove entry");
        }
    } else {
        harness.fail("proxy: upsert entry", "upsert failed");
    }

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(10))
        .build()?;
    if let Err(err) = run_http_tests(&http_base, &client, &mut harness).await {
        harness.fail("http: suite", &format!("{err}"));
    }

    if let Err(err) = run_ipc_tests(mdns.core()?, &mut harness).await {
        harness.fail("ipc: suite", &format!("{err}"));
    }

    http_cancel.cancel();
    handle.shutdown().await?;
    if tokio::time::Instant::now() > total_deadline {
        harness.fail("runtime", "overall timeout exceeded");
    }
    harness.summary();
    if harness.failed > 0 {
        std::process::exit(1);
    }
    Ok(())
}

pub async fn upsert( &self, entry: ProxyEntry, ) -> Result<Vec<ProxyEntry>, KoiError>

Examples found in repository

examples/embedded-integration.rs (line 1022)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args: Vec<String> = std::env::args().collect();
    if has_flag(&args, "--help") {
        println!(
            "Usage: cargo run -p koi-embedded --example embedded-integration -- [options]\n\nOptions:\n  --timeout N   Overall timeout in seconds (default: 30)\n  --verbose     Verbose logging"
        );
        return Ok(());
    }

    let verbose = has_flag(&args, "--verbose");
    let with_certmesh = true;
    let skip_mdns = false;
    let timeout_secs = read_arg_value(&args, "--timeout")
        .and_then(|value| value.parse::<u64>().ok())
        .unwrap_or(30);
    let total_deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);

    let data_dir = temp_data_dir();
    let mut harness = Harness::new(verbose);

    harness.log(format!("data dir: {}", data_dir.display()));
    harness.log(format!("mdns: {}", !skip_mdns));
    harness.log(format!("certmesh: {}", with_certmesh));

    let koi = Builder::new()
        .data_dir(&data_dir)
        .service_mode(ServiceMode::EmbeddedOnly)
        .mdns(!skip_mdns)
        .dns_enabled(true)
        .dns(|cfg| cfg.port(15353))
        .health(true)
        .certmesh(with_certmesh)
        .proxy(true)
        .build()?;
    let handle = koi.start().await?;

    let mdns = match handle.mdns() {
        Ok(mdns) => mdns,
        Err(err) => {
            harness.fail("mdns: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let dns = match handle.dns() {
        Ok(dns) => dns,
        Err(err) => {
            harness.fail("dns: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let health = match handle.health() {
        Ok(health) => health,
        Err(err) => {
            harness.fail("health: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let proxy = match handle.proxy() {
        Ok(proxy) => proxy,
        Err(err) => {
            harness.fail("proxy: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let certmesh = match handle.certmesh() {
        Ok(certmesh) => certmesh,
        Err(err) => {
            harness.fail("certmesh: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };

    let (http_addr, http_cancel) = start_http_server(
        mdns.core()?,
        dns.runtime()?,
        health.core()?,
        certmesh.core()?,
        proxy.runtime()?,
    )
    .await?;
    let http_base = format!("http://{}", http_addr);
    harness.log(format!("http base: {http_base}"));

    // DNS: add entry, lookup, and event.
    let mut rx = handle.subscribe();
    let entry = DnsEntry {
        name: "embedded-test.lan".to_string(),
        ip: "127.0.0.1".to_string(),
        ttl: None,
    };
    let _ = dns.add_entry(entry)?;
    let event = wait_for_event(
        &mut rx,
        Duration::from_secs(2),
        |event| matches!(event, KoiEvent::DnsEntryUpdated { name, .. } if name == "embedded-test.lan"),
    )
    .await;
    if event.is_some() {
        harness.pass("dns: event emitted");
    } else {
        harness.fail("dns: event emitted", "no DnsEntryUpdated event received");
    }

    let result = dns
        .lookup("embedded-test.lan", hickory_proto::rr::RecordType::A)
        .await;
    match result {
        Some(result) => {
            if result.ips.contains(&IpAddr::from([127, 0, 0, 1])) && result.source == "static" {
                harness.pass("dns: lookup static entry");
            } else {
                harness.fail("dns: lookup static entry", "unexpected lookup result");
            }
        }
        None => harness.fail("dns: lookup static entry", "lookup returned none"),
    }

    let names = dns.list_names();
    if names.iter().any(|name| name == "embedded-test.lan.") {
        harness.pass("dns: list names includes entry");
    } else {
        harness.fail("dns: list names includes entry", "name missing from list");
    }

    let _ = dns.remove_entry("embedded-test.lan");
    let removed_event = wait_for_event(
        &mut rx,
        Duration::from_secs(2),
        |event| matches!(event, KoiEvent::DnsEntryRemoved { name } if name == "embedded-test.lan"),
    )
    .await;
    if removed_event.is_some() {
        harness.pass("dns: remove emits removal event");
    } else {
        harness.fail(
            "dns: remove emits removal event",
            "no removal event received",
        );
    }

    // Health: run a TCP check against a local listener.
    let mut rx = handle.subscribe();
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    tokio::spawn(async move {
        loop {
            let _ = listener.accept().await;
        }
    });

    let check = HealthCheck {
        name: "tcp-local".to_string(),
        kind: ServiceCheckKind::Tcp,
        target: format!("127.0.0.1:{}", addr.port()),
        interval_secs: 1,
        timeout_secs: 1,
    };
    health.add_check(check).await?;
    health.core()?.run_checks_once().await;
    let snapshot = health.status().await;
    let status = snapshot
        .services
        .iter()
        .find(|svc| svc.name == "tcp-local")
        .map(|svc| svc.status);
    match status {
        Some(ServiceStatus::Up) => harness.pass("health: tcp check up"),
        Some(other) => harness.fail(
            "health: tcp check up",
            &format!("unexpected status: {other:?}"),
        ),
        None => harness.fail("health: tcp check up", "service missing"),
    }

    let event = wait_for_event(
        &mut rx,
        Duration::from_secs(3),
        |event| matches!(event, KoiEvent::HealthChanged { name, .. } if name == "tcp-local"),
    )
    .await;
    if event.is_some() {
        harness.pass("health: event emitted");
    } else {
        harness.fail("health: event emitted", "no HealthChanged event received");
    }

    let _ = health.remove_check("tcp-local").await;
    let snapshot = health.status().await;
    if snapshot.services.iter().any(|svc| svc.name == "tcp-local") {
        harness.fail("health: remove check", "check still present after removal");
    } else {
        harness.pass("health: remove check");
    }

    // mDNS: register + browse.
    let browse = mdns.browse("_koi._tcp").await;
    if let Ok(browse) = browse {
        let mut txt = HashMap::new();
        txt.insert("source".to_string(), "embedded".to_string());
        let payload = RegisterPayload {
            name: "koi-embedded-test".to_string(),
            service_type: "_koi._tcp".to_string(),
            port: 51515,
            ip: Some("127.0.0.1".to_string()),
            lease_secs: Some(30),
            txt,
        };
        let reg = mdns.register(payload);
        if let Ok(reg) = reg {
            let found = tokio::time::timeout(Duration::from_secs(5), browse.recv()).await;
            match found {
                Ok(Some(_event)) => harness.pass("mdns: register + browse"),
                Ok(None) => harness.fail("mdns: register + browse", "browse stream ended"),
                Err(_) => harness.fail("mdns: register + browse", "no events within timeout"),
            }

            match mdns.resolve("koi-embedded-test._koi._tcp.local.").await {
                Ok(record) if record.port == Some(51515) => {
                    harness.pass("mdns: resolve registered service");
                }
                Ok(_) => harness.fail("mdns: resolve registered service", "unexpected record"),
                Err(err) => harness.fail("mdns: resolve registered service", &format!("{err}")),
            }

            match mdns.unregister(&reg.id) {
                Ok(()) => harness.pass("mdns: unregister"),
                Err(err) => harness.fail("mdns: unregister", &format!("{err}")),
            }
        } else {
            harness.fail("mdns: register + browse", "register failed");
        }
    } else {
        harness.fail("mdns: register + browse", "browse failed");
    }

    // Proxy: upsert and read entries.
    let mut rx = handle.subscribe();
    let entry = ProxyEntry {
        name: "embedded-proxy".to_string(),
        listen_port: 18080,
        backend: "http://127.0.0.1:18081".to_string(),
        allow_remote: false,
    };
    let result = proxy.upsert(entry.clone()).await;
    if result.is_ok() {
        let entries = proxy.entries().await;
        if entries.iter().any(|item| item.name == entry.name) {
            harness.pass("proxy: upsert entry");
        } else {
            harness.fail("proxy: upsert entry", "entry missing after upsert");
        }
        let event = wait_for_event(&mut rx, Duration::from_secs(3), |event| {
            matches!(event, KoiEvent::ProxyEntryUpdated { entry } if entry.name == "embedded-proxy")
        })
        .await;
        if event.is_some() {
            harness.pass("proxy: event emitted");
        } else {
            harness.fail(
                "proxy: event emitted",
                "no ProxyEntryUpdated event received",
            );
        }
        let _ = proxy.remove("embedded-proxy").await;
        let entries = proxy.entries().await;
        if entries.iter().any(|item| item.name == "embedded-proxy") {
            harness.fail("proxy: remove entry", "entry still present after removal");
        } else {
            harness.pass("proxy: remove entry");
        }
    } else {
        harness.fail("proxy: upsert entry", "upsert failed");
    }

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(10))
        .build()?;
    if let Err(err) = run_http_tests(&http_base, &client, &mut harness).await {
        harness.fail("http: suite", &format!("{err}"));
    }

    if let Err(err) = run_ipc_tests(mdns.core()?, &mut harness).await {
        harness.fail("ipc: suite", &format!("{err}"));
    }

    http_cancel.cancel();
    handle.shutdown().await?;
    if tokio::time::Instant::now() > total_deadline {
        harness.fail("runtime", "overall timeout exceeded");
    }
    harness.summary();
    if harness.failed > 0 {
        std::process::exit(1);
    }
    Ok(())
}

pub async fn remove(&self, name: &str) -> Result<Vec<ProxyEntry>, KoiError>

Examples found in repository

examples/embedded-integration.rs (line 1042)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args: Vec<String> = std::env::args().collect();
    if has_flag(&args, "--help") {
        println!(
            "Usage: cargo run -p koi-embedded --example embedded-integration -- [options]\n\nOptions:\n  --timeout N   Overall timeout in seconds (default: 30)\n  --verbose     Verbose logging"
        );
        return Ok(());
    }

    let verbose = has_flag(&args, "--verbose");
    let with_certmesh = true;
    let skip_mdns = false;
    let timeout_secs = read_arg_value(&args, "--timeout")
        .and_then(|value| value.parse::<u64>().ok())
        .unwrap_or(30);
    let total_deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);

    let data_dir = temp_data_dir();
    let mut harness = Harness::new(verbose);

    harness.log(format!("data dir: {}", data_dir.display()));
    harness.log(format!("mdns: {}", !skip_mdns));
    harness.log(format!("certmesh: {}", with_certmesh));

    let koi = Builder::new()
        .data_dir(&data_dir)
        .service_mode(ServiceMode::EmbeddedOnly)
        .mdns(!skip_mdns)
        .dns_enabled(true)
        .dns(|cfg| cfg.port(15353))
        .health(true)
        .certmesh(with_certmesh)
        .proxy(true)
        .build()?;
    let handle = koi.start().await?;

    let mdns = match handle.mdns() {
        Ok(mdns) => mdns,
        Err(err) => {
            harness.fail("mdns: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let dns = match handle.dns() {
        Ok(dns) => dns,
        Err(err) => {
            harness.fail("dns: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let health = match handle.health() {
        Ok(health) => health,
        Err(err) => {
            harness.fail("health: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let proxy = match handle.proxy() {
        Ok(proxy) => proxy,
        Err(err) => {
            harness.fail("proxy: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };
    let certmesh = match handle.certmesh() {
        Ok(certmesh) => certmesh,
        Err(err) => {
            harness.fail("certmesh: setup", &format!("{err}"));
            handle.shutdown().await?;
            harness.summary();
            std::process::exit(1);
        }
    };

    let (http_addr, http_cancel) = start_http_server(
        mdns.core()?,
        dns.runtime()?,
        health.core()?,
        certmesh.core()?,
        proxy.runtime()?,
    )
    .await?;
    let http_base = format!("http://{}", http_addr);
    harness.log(format!("http base: {http_base}"));

    // DNS: add entry, lookup, and event.
    let mut rx = handle.subscribe();
    let entry = DnsEntry {
        name: "embedded-test.lan".to_string(),
        ip: "127.0.0.1".to_string(),
        ttl: None,
    };
    let _ = dns.add_entry(entry)?;
    let event = wait_for_event(
        &mut rx,
        Duration::from_secs(2),
        |event| matches!(event, KoiEvent::DnsEntryUpdated { name, .. } if name == "embedded-test.lan"),
    )
    .await;
    if event.is_some() {
        harness.pass("dns: event emitted");
    } else {
        harness.fail("dns: event emitted", "no DnsEntryUpdated event received");
    }

    let result = dns
        .lookup("embedded-test.lan", hickory_proto::rr::RecordType::A)
        .await;
    match result {
        Some(result) => {
            if result.ips.contains(&IpAddr::from([127, 0, 0, 1])) && result.source == "static" {
                harness.pass("dns: lookup static entry");
            } else {
                harness.fail("dns: lookup static entry", "unexpected lookup result");
            }
        }
        None => harness.fail("dns: lookup static entry", "lookup returned none"),
    }

    let names = dns.list_names();
    if names.iter().any(|name| name == "embedded-test.lan.") {
        harness.pass("dns: list names includes entry");
    } else {
        harness.fail("dns: list names includes entry", "name missing from list");
    }

    let _ = dns.remove_entry("embedded-test.lan");
    let removed_event = wait_for_event(
        &mut rx,
        Duration::from_secs(2),
        |event| matches!(event, KoiEvent::DnsEntryRemoved { name } if name == "embedded-test.lan"),
    )
    .await;
    if removed_event.is_some() {
        harness.pass("dns: remove emits removal event");
    } else {
        harness.fail(
            "dns: remove emits removal event",
            "no removal event received",
        );
    }

    // Health: run a TCP check against a local listener.
    let mut rx = handle.subscribe();
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    tokio::spawn(async move {
        loop {
            let _ = listener.accept().await;
        }
    });

    let check = HealthCheck {
        name: "tcp-local".to_string(),
        kind: ServiceCheckKind::Tcp,
        target: format!("127.0.0.1:{}", addr.port()),
        interval_secs: 1,
        timeout_secs: 1,
    };
    health.add_check(check).await?;
    health.core()?.run_checks_once().await;
    let snapshot = health.status().await;
    let status = snapshot
        .services
        .iter()
        .find(|svc| svc.name == "tcp-local")
        .map(|svc| svc.status);
    match status {
        Some(ServiceStatus::Up) => harness.pass("health: tcp check up"),
        Some(other) => harness.fail(
            "health: tcp check up",
            &format!("unexpected status: {other:?}"),
        ),
        None => harness.fail("health: tcp check up", "service missing"),
    }

    let event = wait_for_event(
        &mut rx,
        Duration::from_secs(3),
        |event| matches!(event, KoiEvent::HealthChanged { name, .. } if name == "tcp-local"),
    )
    .await;
    if event.is_some() {
        harness.pass("health: event emitted");
    } else {
        harness.fail("health: event emitted", "no HealthChanged event received");
    }

    let _ = health.remove_check("tcp-local").await;
    let snapshot = health.status().await;
    if snapshot.services.iter().any(|svc| svc.name == "tcp-local") {
        harness.fail("health: remove check", "check still present after removal");
    } else {
        harness.pass("health: remove check");
    }

    // mDNS: register + browse.
    let browse = mdns.browse("_koi._tcp").await;
    if let Ok(browse) = browse {
        let mut txt = HashMap::new();
        txt.insert("source".to_string(), "embedded".to_string());
        let payload = RegisterPayload {
            name: "koi-embedded-test".to_string(),
            service_type: "_koi._tcp".to_string(),
            port: 51515,
            ip: Some("127.0.0.1".to_string()),
            lease_secs: Some(30),
            txt,
        };
        let reg = mdns.register(payload);
        if let Ok(reg) = reg {
            let found = tokio::time::timeout(Duration::from_secs(5), browse.recv()).await;
            match found {
                Ok(Some(_event)) => harness.pass("mdns: register + browse"),
                Ok(None) => harness.fail("mdns: register + browse", "browse stream ended"),
                Err(_) => harness.fail("mdns: register + browse", "no events within timeout"),
            }

            match mdns.resolve("koi-embedded-test._koi._tcp.local.").await {
                Ok(record) if record.port == Some(51515) => {
                    harness.pass("mdns: resolve registered service");
                }
                Ok(_) => harness.fail("mdns: resolve registered service", "unexpected record"),
                Err(err) => harness.fail("mdns: resolve registered service", &format!("{err}")),
            }

            match mdns.unregister(&reg.id) {
                Ok(()) => harness.pass("mdns: unregister"),
                Err(err) => harness.fail("mdns: unregister", &format!("{err}")),
            }
        } else {
            harness.fail("mdns: register + browse", "register failed");
        }
    } else {
        harness.fail("mdns: register + browse", "browse failed");
    }

    // Proxy: upsert and read entries.
    let mut rx = handle.subscribe();
    let entry = ProxyEntry {
        name: "embedded-proxy".to_string(),
        listen_port: 18080,
        backend: "http://127.0.0.1:18081".to_string(),
        allow_remote: false,
    };
    let result = proxy.upsert(entry.clone()).await;
    if result.is_ok() {
        let entries = proxy.entries().await;
        if entries.iter().any(|item| item.name == entry.name) {
            harness.pass("proxy: upsert entry");
        } else {
            harness.fail("proxy: upsert entry", "entry missing after upsert");
        }
        let event = wait_for_event(&mut rx, Duration::from_secs(3), |event| {
            matches!(event, KoiEvent::ProxyEntryUpdated { entry } if entry.name == "embedded-proxy")
        })
        .await;
        if event.is_some() {
            harness.pass("proxy: event emitted");
        } else {
            harness.fail(
                "proxy: event emitted",
                "no ProxyEntryUpdated event received",
            );
        }
        let _ = proxy.remove("embedded-proxy").await;
        let entries = proxy.entries().await;
        if entries.iter().any(|item| item.name == "embedded-proxy") {
            harness.fail("proxy: remove entry", "entry still present after removal");
        } else {
            harness.pass("proxy: remove entry");
        }
    } else {
        harness.fail("proxy: upsert entry", "upsert failed");
    }

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(10))
        .build()?;
    if let Err(err) = run_http_tests(&http_base, &client, &mut harness).await {
        harness.fail("http: suite", &format!("{err}"));
    }

    if let Err(err) = run_ipc_tests(mdns.core()?, &mut harness).await {
        harness.fail("ipc: suite", &format!("{err}"));
    }

    http_cancel.cancel();
    handle.shutdown().await?;
    if tokio::time::Instant::now() > total_deadline {
        harness.fail("runtime", "overall timeout exceeded");
    }
    harness.summary();
    if harness.failed > 0 {
        std::process::exit(1);
    }
    Ok(())
}

pub async fn start_all(&self) -> Result<(), KoiError>

pub async fn stop_all(&self)