use anyhow::Result;
use clap::Parser;
use std::net::SocketAddr;
use tokio::sync::mpsc;
use tracing::debug;
use crate::client::ManagementResponse;
use crate::tui;
use crate::utils::event_display::{display_events_header, display_single_event};
use crate::{error::CliError, output::formatters::ActorStarted, CommandContext};
use theater::utils::resolve_reference;
#[derive(Debug, Parser)]
pub struct StartArgs {
#[arg(required = true)]
pub manifest: String,
#[arg(short, long)]
pub address: Option<SocketAddr>,
#[arg(short, long)]
pub initial_state: Option<String>,
#[arg(short, long)]
pub subscribe: bool,
#[arg(short, long)]
pub parent: bool,
#[arg(long)]
pub id_only: bool,
#[arg(short, long, default_value = "compact")]
pub format: String,
}
pub async fn execute_async(args: &StartArgs, ctx: &CommandContext) -> Result<(), CliError> {
debug!("Starting actor from manifest: {}", args.manifest);
let address = ctx.server_address(args.address);
debug!("Connecting to server at: {}", address);
let manifest_bytes = resolve_reference(&args.manifest).await.map_err(|e| {
CliError::invalid_manifest(format!(
"Failed to resolve manifest reference '{}': {}",
args.manifest, e
))
})?;
let manifest_content = String::from_utf8(manifest_bytes).map_err(|e| {
CliError::invalid_manifest(format!("Manifest content is not valid UTF-8: {}", e))
})?;
let initial_state = if let Some(state_str) = &args.initial_state {
match resolve_reference(state_str).await {
Ok(bytes) => {
debug!("Resolved initial state from reference: {}", state_str);
Some(bytes)
}
Err(_) => {
debug!("Using provided string as JSON initial state");
Some(state_str.as_bytes().to_vec())
}
}
} else {
None
};
let client = ctx.create_client();
client
.connect()
.await
.map_err(|e| CliError::connection_failed(address, e))?;
debug!("Calling start actor on client");
client
.start_actor(manifest_content, initial_state, args.parent, args.subscribe)
.await
.map_err(|e| CliError::actor_not_found(format!("Failed to start actor: {}", e)))?;
debug!("Actor start request sent successfully");
let use_tui = args.subscribe && args.parent && !ctx.json && !args.id_only;
if use_tui {
return run_with_tui(args, client).await;
} else if args.subscribe && !ctx.json {
println!("");
display_events_header(&args.format);
}
let mut actor_started = false;
let timeout_duration = tokio::time::Duration::from_secs(30);
debug!("Entering response loop, waiting for actor start confirmation or events");
loop {
tokio::select! {
data = client.next_response() => {
debug!("Received response from client");
debug!("Response data: {:?}", data);
if let Ok(data) = data {
match data {
ManagementResponse::ActorStarted { id } => {
debug!("Management response received: Actor started with ID: {}", id);
actor_started = true;
if args.id_only {
println!("{}", id);
break;
} else {
let result = ActorStarted {
actor_id: id.to_string(),
manifest_path: args.manifest.clone(),
address: address.to_string(),
subscribing: args.subscribe,
acting_as_parent: args.parent,
};
debug!("Outputting result: {:?}", result);
ctx.output.output(&result, None)?;
if !args.subscribe && !args.parent {
break;
}
}
}
ManagementResponse::ActorEvent { event } => {
if args.subscribe {
display_single_event(&event, &args.format)
.map_err(|e| CliError::invalid_input("event_display", "event", e.to_string()))?;
}
}
ManagementResponse::ActorError { error } => {
if args.subscribe {
println!("-----[actor error]-----------------");
println!(" {}", error);
println!("-----------------------------------");
}
}
ManagementResponse::ActorStopped { id } => {
println!("-----[actor stopped]-----------------");
println!("{}", id);
println!("-------------------------------------");
break;
}
ManagementResponse::ActorResult(actor_result) => {
if args.parent {
println!("-----[actor result]-----------------");
println!(" {}", actor_result);
println!("------------------------------------");
}
}
ManagementResponse::Error { error } => {
return Err(CliError::management_error(error));
}
_ => {
println!("Unknown response received");
break;
}
}
}
}
_ = tokio::time::sleep(timeout_duration) => {
if !actor_started {
return Err(CliError::operation_timeout("Actor startup", timeout_duration.as_secs()));
}
}
_ = tokio::signal::ctrl_c() => {
debug!("Received Ctrl-C, stopping");
if !ctx.json {
println!("\n{}\n", "Interrupted by user");
}
break;
}
}
}
Ok(())
}
async fn run_with_tui(
args: &StartArgs,
client: crate::client::TheaterClient,
) -> Result<(), CliError> {
debug!("Starting TUI mode for actor monitoring");
let (response_tx, response_rx) = mpsc::unbounded_channel();
let mut _actor_id: Option<String> = None;
let mut tui_started = false;
let mut tui_completed = false;
let timeout_duration = tokio::time::Duration::from_secs(30);
let mut tui_handle = {
let manifest_path = args.manifest.clone();
tokio::spawn(async move {
if let Err(e) =
tui::run_tui("Starting...".to_string(), manifest_path, response_rx).await
{
eprintln!("TUI error: {}", e);
}
})
};
loop {
tokio::select! {
data = client.next_response() => {
if let Ok(response) = data {
match &response {
ManagementResponse::ActorStarted { id } => {
_actor_id = Some(id.to_string());
debug!("Actor started with ID: {}", id);
tui_started = true;
}
ManagementResponse::ActorStopped { .. } => {
let _ = response_tx.send(response);
break;
}
_ => {}
}
if let Err(_) = response_tx.send(response) {
debug!("TUI channel closed, stopping");
break;
}
}
}
_ = tokio::time::sleep(timeout_duration) => {
if !tui_started {
return Err(CliError::operation_timeout("Actor startup", timeout_duration.as_secs()));
}
}
_ = tokio::signal::ctrl_c() => {
debug!("Received Ctrl-C, stopping TUI mode");
break;
}
result = &mut tui_handle, if !tui_completed => {
match result {
Ok(_) => debug!("TUI task completed"),
Err(e) => debug!("TUI task error: {}", e),
}
tui_completed = true;
break;
}
}
}
if !tui_completed {
let _ = tokio::time::timeout(tokio::time::Duration::from_millis(500), tui_handle).await;
}
Ok(())
}