Split cli::run() into named startup-phase functions

run() bundled settings loading, engine spawn, active-profile resolve/load,
ingest spawn, and session registration inline. Extract each into a named
function (spawn_engine, load_active_profile, spawn_ingest,
register_session_and_log) so run() reads as a sequence of steps. No
behavior change.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ
This commit is contained in:
Hugh Kaznowski
2026-09-07 10:28:51 +01:00
co-authored by Claude Sonnet 5
parent f96de8f38d
commit 1b6c50075a
+59 -41
View File
@@ -71,38 +71,67 @@ pub async fn run() {
let settings = Arc::new(Mutex::new(Settings::load(&settings_path))); let settings = Arc::new(Mutex::new(Settings::load(&settings_path)));
let engine_tx = spawn_engine(&settings).await;
load_active_profile(&settings, &settings_path, &engine_tx).await;
spawn_ingest(&args, &engine_tx);
let app = crate::api::router(engine_tx, settings, settings_path.clone());
let listener = TcpListener::bind((crate::consts::DEFAULT_BIND_HOST, args.port))
.await
.expect("failed to bind listener");
let bound_port = listener
.local_addr()
.expect("failed to read back bound address")
.port();
register_session_and_log(&settings_path, &args, bound_port);
axum::serve(listener, app)
.await
.expect("server error");
}
async fn spawn_engine(settings: &Arc<Mutex<Settings>>) -> crate::engine::EngineHandle {
let engine_tx = crate::engine::spawn(settings.clone(), crate::consts::ENGINE_CHANNEL_CAPACITY); let engine_tx = crate::engine::spawn(settings.clone(), crate::consts::ENGINE_CHANNEL_CAPACITY);
{ let max_logs = settings.lock().unwrap().max_logs;
let max_logs = settings.lock().unwrap().max_logs; let _ = engine_tx.tx.send(crate::engine::EngineMessage::SetMaxLogs(max_logs)).await;
let _ = engine_tx.tx.send(crate::engine::EngineMessage::SetMaxLogs(max_logs)).await; engine_tx
} }
{ /// Resolves the active profile (creating the Default profile if none exists yet) and, if
let (active_profile_name, active_profile) = { /// found, replays its declarations into the engine before ingest starts — there's no way to
let mut settings_guard = settings.lock().unwrap(); /// reprocess earlier lines later, so this must happen before `spawn_ingest`.
settings_guard async fn load_active_profile(
.ensure_default_profile(&settings_path) settings: &Arc<Mutex<Settings>>,
.expect("failed to ensure the Default profile exists"); settings_path: &std::path::Path,
let active_profile_name = settings_guard.active_profile.clone(); engine_tx: &crate::engine::EngineHandle,
let active_profile = active_profile_name ) {
.as_deref() let (active_profile_name, active_profile) = {
.and_then(|name| settings_guard.profile(name).cloned()); let mut settings_guard = settings.lock().unwrap();
(active_profile_name, active_profile) settings_guard
}; .ensure_default_profile(settings_path)
match (active_profile_name, active_profile) { .expect("failed to ensure the Default profile exists");
(Some(_), Some(active_profile)) => { let active_profile_name = settings_guard.active_profile.clone();
active_profile.load_into_engine(&engine_tx).await; let active_profile = active_profile_name
} .as_deref()
(Some(profile_name), None) => { .and_then(|name| settings_guard.profile(name).cloned());
eprintln!( (active_profile_name, active_profile)
"warning: active profile '{profile_name}' not found in {}; skipping", };
settings_path.display() match (active_profile_name, active_profile) {
); (Some(_), Some(active_profile)) => {
} active_profile.load_into_engine(engine_tx).await;
(None, _) => {}
} }
(Some(profile_name), None) => {
eprintln!(
"warning: active profile '{profile_name}' not found in {}; skipping",
settings_path.display()
);
}
(None, _) => {}
} }
}
fn spawn_ingest(args: &Args, engine_tx: &crate::engine::EngineHandle) {
match args.file.clone() { match args.file.clone() {
Some(path) => { Some(path) => {
tokio::spawn(crate::engine::ingest::run_file(engine_tx.clone(), path)); tokio::spawn(crate::engine::ingest::run_file(engine_tx.clone(), path));
@@ -115,22 +144,15 @@ pub async fn run() {
if let Some(metrics_url) = args.metrics_url.clone() { if let Some(metrics_url) = args.metrics_url.clone() {
tokio::spawn(crate::engine::ingest::run_metrics_poll(engine_tx.clone(), metrics_url)); tokio::spawn(crate::engine::ingest::run_metrics_poll(engine_tx.clone(), metrics_url));
} }
}
let app = crate::api::router(engine_tx, settings, settings_path.clone()); fn register_session_and_log(settings_path: &std::path::Path, args: &Args, bound_port: u16) {
let listener = TcpListener::bind((crate::consts::DEFAULT_BIND_HOST, args.port))
.await
.expect("failed to bind listener");
let bound_port = listener
.local_addr()
.expect("failed to read back bound address")
.port();
let source = match &args.file { let source = match &args.file {
Some(path) => format!("tailing {}", path.display()), Some(path) => format!("tailing {}", path.display()),
None => "reading logs from stdin".to_string(), None => "reading logs from stdin".to_string(),
}; };
let session_name = args.name.clone().unwrap_or_else(sessions::generate_name); let session_name = args.name.clone().unwrap_or_else(sessions::generate_name);
match sessions::register_session(&settings_path, &session_name, bound_port) { match sessions::register_session(settings_path, &session_name, bound_port) {
Ok(_) => eprintln!( Ok(_) => eprintln!(
"ottyl serving on http://localhost:{} ({}), session name: {}", "ottyl serving on http://localhost:{} ({}), session name: {}",
bound_port, source, session_name bound_port, source, session_name
@@ -140,10 +162,6 @@ pub async fn run() {
bound_port, source, session_name, err bound_port, source, session_name, err
), ),
} }
axum::serve(listener, app)
.await
.expect("server error");
} }
async fn run_mcp(settings_path: &std::path::Path, session: Option<String>, port: Option<u16>) { async fn run_mcp(settings_path: &std::path::Path, session: Option<String>, port: Option<u16>) {