1use crate::server::McpServer;
7use rustapi_core::RustApi;
8use std::error::Error;
9use std::future::Future;
10use std::pin::Pin;
11use tokio::sync::watch;
12
13pub type BoxError = Box<dyn Error + Send + Sync>;
15
16pub type Result<T> = std::result::Result<T, BoxError>;
18
19pub type ShutdownFuture = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
21
22fn to_boxed_error<E>(err: E) -> BoxError
23where
24 E: Error + Send + Sync + 'static,
25{
26 Box::new(err)
27}
28
29pub async fn run_concurrently<HF, MF, HE, ME>(http_future: HF, mcp_future: MF) -> Result<()>
31where
32 HF: Future<Output = std::result::Result<(), HE>> + Send,
33 MF: Future<Output = std::result::Result<(), ME>> + Send,
34 HE: Error + Send + Sync + 'static,
35 ME: Error + Send + Sync + 'static,
36{
37 let http_task = async move { http_future.await.map_err(to_boxed_error) };
38 let mcp_task = async move { mcp_future.await.map_err(to_boxed_error) };
39
40 let (_http_ok, _mcp_ok) = tokio::try_join!(http_task, mcp_task)?;
41 Ok(())
42}
43
44pub async fn run_rustapi_and_mcp(
49 app: RustApi,
50 http_addr: impl AsRef<str>,
51 mcp: McpServer,
52 mcp_addr: impl AsRef<str>,
53) -> Result<()>
54where
55{
56 let http_addr = http_addr.as_ref().to_string();
57 let mcp_addr = mcp_addr.as_ref().to_string();
58
59 let http_base = format!(
62 "http://127.0.0.1:{}",
63 extract_port_or_default(&http_addr, 8080)
64 );
65 let mcp = mcp.with_http_base(http_base);
66
67 let http_task = async move { app.run(&http_addr).await };
68 let mcp_task = async move { mcp.serve(&mcp_addr).await.map_err(to_boxed_error) };
69
70 let (_http_ok, _mcp_ok) = tokio::try_join!(http_task, mcp_task)?;
71 Ok(())
72}
73
74pub async fn run_rustapi_and_mcp_with_shutdown<SF>(
78 app: RustApi,
79 http_addr: impl AsRef<str>,
80 mcp: McpServer,
81 mcp_addr: impl AsRef<str>,
82 shutdown_signal: SF,
83) -> Result<()>
84where
85 SF: Future<Output = ()> + Send + 'static,
86{
87 let http_addr = http_addr.as_ref().to_string();
88 let mcp_addr = mcp_addr.as_ref().to_string();
89
90 let http_base = format!(
91 "http://127.0.0.1:{}",
92 extract_port_or_default(&http_addr, 8080)
93 );
94 let mcp = mcp.with_http_base(http_base);
95
96 let (shutdown_tx, shutdown_rx) = watch::channel(false);
97
98 let shutdown_dispatch = tokio::spawn(async move {
99 shutdown_signal.await;
100 let _ = shutdown_tx.send(true);
101 });
102
103 let http_shutdown = shutdown_notifier(shutdown_rx.clone());
104 let mcp_shutdown = shutdown_notifier(shutdown_rx);
105
106 let http_task = async move { app.run_with_shutdown(&http_addr, http_shutdown).await };
107
108 let mcp_task = async move {
109 mcp.serve_with_shutdown(&mcp_addr, mcp_shutdown)
110 .await
111 .map_err(to_boxed_error)
112 };
113
114 let joined = tokio::try_join!(http_task, mcp_task).map(|_| ());
115
116 shutdown_dispatch.abort();
117 let _ = shutdown_dispatch.await;
118
119 joined
120}
121
122async fn shutdown_notifier(mut rx: watch::Receiver<bool>) {
123 if *rx.borrow() {
124 return;
125 }
126 while rx.changed().await.is_ok() {
127 if *rx.borrow() {
128 break;
129 }
130 }
131}
132
133fn extract_port_or_default(addr: &str, default: u16) -> u16 {
135 if let Some(colon) = addr.rfind(':') {
137 let after = &addr[colon + 1..];
138 let port_str = after
140 .split(|c: char| !c.is_ascii_digit())
141 .next()
142 .unwrap_or("");
143 if let Ok(p) = port_str.parse::<u16>() {
144 return p;
145 }
146 }
147 default
148}