Function exec

crates/cli/src/subcommands/subscribe.rs:151–235

(config: Config, args: &ArgMatches)

}

pub async fn exec(config: Config, args: &ArgMatches) -> Result<(), anyhow::Error> {
    eprintln!("{UNSTABLE_WARNING}\n");

    let server = args.get_one::<String>("server").map(|s| s.as_ref());
    let force = args.get_flag("force");
    let anon_identity = args.get_flag("anon_identity");
    let no_config = args.get_flag("no_config");

    let raw_parts: Vec<String> = args
        .get_many::<String>("subscribe_parts")
        .map(|vals| vals.cloned().collect())
        .unwrap_or_default();
    let config_targets = load_config_db_targets(no_config)?;
    let resolved = resolve_optional_database_parts(
        &raw_parts,
        config_targets.as_deref(),
        "query",
        "spacetime subscribe [database] <query> [query...] (or --no-config for legacy behavior)",
    )?;
    let queries: Vec<String> = resolved.remaining_args;

    let num = args.get_one::<u32>("num-updates").copied();
    let timeout = args.get_one::<u32>("timeout").copied();
    let print_initial_update = args.get_flag("print_initial_update");
    let confirmed = args.get_one::<bool>("confirmed").copied();
    let resolved_server = server.or(resolved.server.as_deref());

    let mut config = config;
    let conn = crate::api::Connection {
        host: config.get_host_url(resolved_server)?,
        auth_header: get_auth_header(&mut config, anon_identity, resolved_server, !force).await?,
        database_identity: database_identity(&config, &resolved.database, resolved_server).await?,
        database: resolved.database.clone(),
    };
    let api = ClientApi::new(conn);
    let module_def = api.module_def().await?;
    let mut conn = connect_with_fallback(&api, confirmed).await?;
    let num_received = UpdateCounter::new();

    let task = async {
        conn.subscribe(queries.iter().cloned().map(Into::into).collect())
            .await?;
        conn.await_initial_update(print_initial_update.then_some(&module_def))
            .await?;
        conn.consume_transaction_updates(num, &module_def, &num_received).await
    };

    let res = if let Some(timeout) = timeout {
        let timeout = Duration::from_secs(timeout.into());
        match tokio::time::timeout(timeout, task).await {
            Ok(res) => res,
            Err(_elapsed) => {
                let received = num_received.get();
                eprintln!("timed out after {}s", timeout.as_secs());
                match num {
                    Some(expected) if received < expected => Err(Error::UpdateLimitTimedOut {
                        expected,
                        received,