lib.rs 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717
  1. mod cli;
  2. mod constants;
  3. #[cfg(target_os = "linux")]
  4. pub mod linux_display;
  5. #[cfg(target_os = "linux")]
  6. pub mod linux_windowing;
  7. mod logging;
  8. mod markdown;
  9. mod os;
  10. mod server;
  11. mod window_customizer;
  12. mod windows;
  13. use crate::cli::CommandChild;
  14. use futures::{
  15. FutureExt, TryFutureExt,
  16. future::{self, Shared},
  17. };
  18. use std::{
  19. env,
  20. net::TcpListener,
  21. path::PathBuf,
  22. process::Command,
  23. sync::{Arc, Mutex},
  24. time::Duration,
  25. };
  26. use tauri::{AppHandle, Listener, Manager, RunEvent, State, ipc::Channel};
  27. #[cfg(any(target_os = "linux", all(debug_assertions, windows)))]
  28. use tauri_plugin_deep_link::DeepLinkExt;
  29. use tauri_specta::Event;
  30. use tokio::{
  31. sync::{oneshot, watch},
  32. time::{sleep, timeout},
  33. };
  34. use crate::cli::{sqlite_migration::SqliteMigrationProgress, sync_cli};
  35. use crate::constants::*;
  36. use crate::server::get_saved_server_url;
  37. use crate::windows::{LoadingWindow, MainWindow};
  38. #[derive(Clone, serde::Serialize, specta::Type, Debug)]
  39. struct ServerReadyData {
  40. url: String,
  41. username: Option<String>,
  42. password: Option<String>,
  43. is_sidecar: bool,
  44. }
  45. #[derive(Clone, Copy, serde::Serialize, specta::Type, Debug)]
  46. #[serde(tag = "phase", rename_all = "snake_case")]
  47. enum InitStep {
  48. ServerWaiting,
  49. SqliteWaiting,
  50. Done,
  51. }
  52. #[derive(serde::Deserialize, specta::Type)]
  53. #[serde(rename_all = "snake_case")]
  54. enum WslPathMode {
  55. Windows,
  56. Linux,
  57. }
  58. struct InitState {
  59. current: watch::Receiver<InitStep>,
  60. }
  61. #[derive(Clone)]
  62. struct ServerState {
  63. child: Arc<Mutex<Option<CommandChild>>>,
  64. status: future::Shared<oneshot::Receiver<Result<ServerReadyData, String>>>,
  65. }
  66. impl ServerState {
  67. pub fn new(
  68. child: Option<CommandChild>,
  69. status: Shared<oneshot::Receiver<Result<ServerReadyData, String>>>,
  70. ) -> Self {
  71. Self {
  72. child: Arc::new(Mutex::new(child)),
  73. status,
  74. }
  75. }
  76. pub fn set_child(&self, child: Option<CommandChild>) {
  77. *self.child.lock().unwrap() = child;
  78. }
  79. }
  80. #[tauri::command]
  81. #[specta::specta]
  82. fn kill_sidecar(app: AppHandle) {
  83. let Some(server_state) = app.try_state::<ServerState>() else {
  84. tracing::info!("Server not running");
  85. return;
  86. };
  87. let Some(server_state) = server_state
  88. .child
  89. .lock()
  90. .expect("Failed to acquire mutex lock")
  91. .take()
  92. else {
  93. tracing::info!("Server state missing");
  94. return;
  95. };
  96. let _ = server_state.kill();
  97. tracing::info!("Killed server");
  98. }
  99. fn get_logs() -> String {
  100. logging::tail()
  101. }
  102. #[tauri::command]
  103. #[specta::specta]
  104. async fn await_initialization(
  105. state: State<'_, ServerState>,
  106. init_state: State<'_, InitState>,
  107. events: Channel<InitStep>,
  108. ) -> Result<ServerReadyData, String> {
  109. let mut rx = init_state.current.clone();
  110. let events = async {
  111. let e = *rx.borrow();
  112. let _ = events.send(e);
  113. while rx.changed().await.is_ok() {
  114. let step = *rx.borrow_and_update();
  115. let _ = events.send(step);
  116. if matches!(step, InitStep::Done) {
  117. break;
  118. }
  119. }
  120. };
  121. future::join(state.status.clone(), events)
  122. .await
  123. .0
  124. .map_err(|_| "Failed to get server status".to_string())?
  125. }
  126. #[tauri::command]
  127. #[specta::specta]
  128. fn check_app_exists(app_name: &str) -> bool {
  129. #[cfg(target_os = "windows")]
  130. {
  131. os::windows::check_windows_app(app_name)
  132. }
  133. #[cfg(target_os = "macos")]
  134. {
  135. check_macos_app(app_name)
  136. }
  137. #[cfg(target_os = "linux")]
  138. {
  139. check_linux_app(app_name)
  140. }
  141. }
  142. #[tauri::command]
  143. #[specta::specta]
  144. fn resolve_app_path(app_name: &str) -> Option<String> {
  145. #[cfg(target_os = "windows")]
  146. {
  147. os::windows::resolve_windows_app_path(app_name)
  148. }
  149. #[cfg(not(target_os = "windows"))]
  150. {
  151. // On macOS/Linux, just return the app_name as-is since
  152. // the opener plugin handles them correctly
  153. Some(app_name.to_string())
  154. }
  155. }
  156. #[tauri::command]
  157. #[specta::specta]
  158. fn open_path(_app: AppHandle, path: String, app_name: Option<String>) -> Result<(), String> {
  159. #[cfg(target_os = "windows")]
  160. {
  161. let app_name = app_name.map(|v| os::windows::resolve_windows_app_path(&v).unwrap_or(v));
  162. let is_powershell = app_name.as_ref().is_some_and(|v| {
  163. std::path::Path::new(v)
  164. .file_name()
  165. .and_then(|name| name.to_str())
  166. .is_some_and(|name| {
  167. name.eq_ignore_ascii_case("powershell")
  168. || name.eq_ignore_ascii_case("powershell.exe")
  169. })
  170. });
  171. if is_powershell {
  172. return os::windows::open_in_powershell(path);
  173. }
  174. return tauri_plugin_opener::open_path(path, app_name.as_deref())
  175. .map_err(|e| format!("Failed to open path: {e}"));
  176. }
  177. #[cfg(not(target_os = "windows"))]
  178. tauri_plugin_opener::open_path(path, app_name.as_deref())
  179. .map_err(|e| format!("Failed to open path: {e}"))
  180. }
  181. #[cfg(target_os = "macos")]
  182. fn check_macos_app(app_name: &str) -> bool {
  183. // Check common installation locations
  184. let mut app_locations = vec![
  185. format!("/Applications/{}.app", app_name),
  186. format!("/System/Applications/{}.app", app_name),
  187. ];
  188. if let Ok(home) = std::env::var("HOME") {
  189. app_locations.push(format!("{}/Applications/{}.app", home, app_name));
  190. }
  191. for location in app_locations {
  192. if std::path::Path::new(&location).exists() {
  193. return true;
  194. }
  195. }
  196. // Also check if command exists in PATH
  197. Command::new("which")
  198. .arg(app_name)
  199. .output()
  200. .map(|output| output.status.success())
  201. .unwrap_or(false)
  202. }
  203. #[derive(serde::Serialize, serde::Deserialize, specta::Type)]
  204. #[serde(rename_all = "camelCase")]
  205. pub enum LinuxDisplayBackend {
  206. Wayland,
  207. Auto,
  208. }
  209. #[tauri::command]
  210. #[specta::specta]
  211. fn get_display_backend() -> Option<LinuxDisplayBackend> {
  212. #[cfg(target_os = "linux")]
  213. {
  214. let prefer = linux_display::read_wayland().unwrap_or(false);
  215. return Some(if prefer {
  216. LinuxDisplayBackend::Wayland
  217. } else {
  218. LinuxDisplayBackend::Auto
  219. });
  220. }
  221. #[cfg(not(target_os = "linux"))]
  222. None
  223. }
  224. #[tauri::command]
  225. #[specta::specta]
  226. fn set_display_backend(_app: AppHandle, _backend: LinuxDisplayBackend) -> Result<(), String> {
  227. #[cfg(target_os = "linux")]
  228. {
  229. let prefer = matches!(_backend, LinuxDisplayBackend::Wayland);
  230. return linux_display::write_wayland(&_app, prefer);
  231. }
  232. #[cfg(not(target_os = "linux"))]
  233. Ok(())
  234. }
  235. #[cfg(target_os = "linux")]
  236. fn check_linux_app(app_name: &str) -> bool {
  237. return true;
  238. }
  239. #[tauri::command]
  240. #[specta::specta]
  241. fn wsl_path(path: String, mode: Option<WslPathMode>) -> Result<String, String> {
  242. if !cfg!(windows) {
  243. return Ok(path);
  244. }
  245. let flag = match mode.unwrap_or(WslPathMode::Linux) {
  246. WslPathMode::Windows => "-w",
  247. WslPathMode::Linux => "-u",
  248. };
  249. let output = if path.starts_with('~') {
  250. let suffix = path.strip_prefix('~').unwrap_or("");
  251. let escaped = suffix.replace('"', "\\\"");
  252. let cmd = format!("wslpath {flag} \"$HOME{escaped}\"");
  253. Command::new("wsl")
  254. .args(["-e", "sh", "-lc", &cmd])
  255. .output()
  256. .map_err(|e| format!("Failed to run wslpath: {e}"))?
  257. } else {
  258. Command::new("wsl")
  259. .args(["-e", "wslpath", flag, &path])
  260. .output()
  261. .map_err(|e| format!("Failed to run wslpath: {e}"))?
  262. };
  263. if !output.status.success() {
  264. let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
  265. if stderr.is_empty() {
  266. return Err("wslpath failed".to_string());
  267. }
  268. return Err(stderr);
  269. }
  270. Ok(String::from_utf8_lossy(&output.stdout).trim().to_string())
  271. }
  272. #[cfg_attr(mobile, tauri::mobile_entry_point)]
  273. pub fn run() {
  274. let builder = make_specta_builder();
  275. #[cfg(debug_assertions)] // <- Only export on non-release builds
  276. export_types(&builder);
  277. #[cfg(all(target_os = "macos", not(debug_assertions)))]
  278. let _ = std::process::Command::new("killall")
  279. .arg("opencode-cli")
  280. .output();
  281. let mut builder = tauri::Builder::default()
  282. .plugin(tauri_plugin_single_instance::init(|app, _args, _cwd| {
  283. // Focus existing window when another instance is launched
  284. if let Some(window) = app.get_webview_window(MainWindow::LABEL) {
  285. let _ = window.set_focus();
  286. let _ = window.unminimize();
  287. }
  288. }))
  289. .plugin(tauri_plugin_deep_link::init())
  290. .plugin(tauri_plugin_os::init())
  291. .plugin(
  292. tauri_plugin_window_state::Builder::new()
  293. .with_state_flags(window_state_flags())
  294. .with_denylist(&[LoadingWindow::LABEL])
  295. .build(),
  296. )
  297. .plugin(tauri_plugin_store::Builder::new().build())
  298. .plugin(tauri_plugin_dialog::init())
  299. .plugin(tauri_plugin_shell::init())
  300. .plugin(tauri_plugin_process::init())
  301. .plugin(tauri_plugin_opener::init())
  302. .plugin(tauri_plugin_clipboard_manager::init())
  303. .plugin(tauri_plugin_http::init())
  304. .plugin(tauri_plugin_notification::init())
  305. .plugin(crate::window_customizer::PinchZoomDisablePlugin)
  306. .plugin(tauri_plugin_decorum::init())
  307. .invoke_handler(builder.invoke_handler())
  308. .setup(move |app| {
  309. let handle = app.handle().clone();
  310. let log_dir = app
  311. .path()
  312. .app_log_dir()
  313. .expect("failed to resolve app log dir");
  314. // Hold the guard in managed state so it lives for the app's lifetime,
  315. // ensuring all buffered logs are flushed on shutdown.
  316. handle.manage(logging::init(&log_dir));
  317. builder.mount_events(&handle);
  318. tauri::async_runtime::spawn(initialize(handle));
  319. Ok(())
  320. });
  321. if UPDATER_ENABLED {
  322. builder = builder.plugin(tauri_plugin_updater::Builder::new().build());
  323. }
  324. builder
  325. .build(tauri::generate_context!())
  326. .expect("error while running tauri application")
  327. .run(|app, event| {
  328. if let RunEvent::Exit = event {
  329. tracing::info!("Received Exit");
  330. kill_sidecar(app.clone());
  331. }
  332. });
  333. }
  334. fn make_specta_builder() -> tauri_specta::Builder<tauri::Wry> {
  335. tauri_specta::Builder::<tauri::Wry>::new()
  336. // Then register them (separated by a comma)
  337. .commands(tauri_specta::collect_commands![
  338. kill_sidecar,
  339. cli::install_cli,
  340. await_initialization,
  341. server::get_default_server_url,
  342. server::set_default_server_url,
  343. server::get_wsl_config,
  344. server::set_wsl_config,
  345. get_display_backend,
  346. set_display_backend,
  347. markdown::parse_markdown_command,
  348. check_app_exists,
  349. wsl_path,
  350. resolve_app_path,
  351. open_path
  352. ])
  353. .events(tauri_specta::collect_events![
  354. LoadingWindowComplete,
  355. SqliteMigrationProgress
  356. ])
  357. .error_handling(tauri_specta::ErrorHandlingMode::Throw)
  358. }
  359. fn export_types(builder: &tauri_specta::Builder<tauri::Wry>) {
  360. builder
  361. .export(
  362. specta_typescript::Typescript::default(),
  363. "../src/bindings.ts",
  364. )
  365. .expect("Failed to export typescript bindings");
  366. }
  367. #[cfg(test)]
  368. #[test]
  369. fn test_export_types() {
  370. let builder = make_specta_builder();
  371. export_types(&builder);
  372. }
  373. #[derive(tauri_specta::Event, serde::Deserialize, specta::Type)]
  374. struct LoadingWindowComplete;
  375. async fn initialize(app: AppHandle) {
  376. tracing::info!("Initializing app");
  377. let (init_tx, init_rx) = watch::channel(InitStep::ServerWaiting);
  378. setup_app(&app, init_rx);
  379. spawn_cli_sync_task(app.clone());
  380. let (server_ready_tx, server_ready_rx) = oneshot::channel();
  381. let server_ready_rx = server_ready_rx.shared();
  382. app.manage(ServerState::new(None, server_ready_rx.clone()));
  383. let loading_window_complete = event_once_fut::<LoadingWindowComplete>(&app);
  384. tracing::info!("Main and loading windows created");
  385. // SQLite migration handling:
  386. // We only do this if the sqlite db doesn't exist, and we're expecting the sidecar to create it
  387. // First, we spawn a task that listens for SqliteMigrationProgress events that can
  388. // come from any invocation of the sidecar CLI. The progress is captured by a stdout stream interceptor.
  389. // Then in the loading task, we wait for sqlite migration to complete before
  390. // starting our health check against the server, otherwise long migrations could result in a timeout.
  391. let needs_sqlite_migration = !sqlite_file_exists();
  392. let sqlite_done = needs_sqlite_migration.then(|| {
  393. tracing::info!(
  394. path = %opencode_db_path().expect("failed to get db path").display(),
  395. "Sqlite file not found, waiting for it to be generated"
  396. );
  397. let (done_tx, done_rx) = oneshot::channel::<()>();
  398. let done_tx = Arc::new(Mutex::new(Some(done_tx)));
  399. let init_tx = init_tx.clone();
  400. let id = SqliteMigrationProgress::listen(&app, move |e| {
  401. let _ = init_tx.send(InitStep::SqliteWaiting);
  402. if matches!(e.payload, SqliteMigrationProgress::Done)
  403. && let Some(done_tx) = done_tx.lock().unwrap().take()
  404. {
  405. let _ = done_tx.send(());
  406. }
  407. });
  408. let app = app.clone();
  409. tokio::spawn(done_rx.map(async move |_| {
  410. app.unlisten(id);
  411. }))
  412. });
  413. let loading_task = tokio::spawn({
  414. let app = app.clone();
  415. async move {
  416. tracing::info!("Setting up server connection");
  417. let server_connection = setup_server_connection(app.clone()).await;
  418. tracing::info!("Server connection setup");
  419. // we delay spawning this future so that the timeout is created lazily
  420. let cli_health_check = match server_connection {
  421. ServerConnection::CLI {
  422. child,
  423. health_check,
  424. url,
  425. username,
  426. password,
  427. } => {
  428. let app = app.clone();
  429. Some(
  430. async move {
  431. let res = timeout(Duration::from_secs(30), health_check.0).await;
  432. let err = match res {
  433. Ok(Ok(Ok(()))) => None,
  434. Ok(Ok(Err(e))) => Some(e),
  435. Ok(Err(e)) => Some(format!("Health check task failed: {e}")),
  436. Err(_) => Some("Health check timed out".to_string()),
  437. };
  438. if let Some(err) = err {
  439. let _ = child.kill();
  440. return Err(format!(
  441. "Failed to spawn OpenCode Server ({err}). Logs:\n{}",
  442. get_logs()
  443. ));
  444. }
  445. tracing::info!("CLI health check OK");
  446. app.state::<ServerState>().set_child(Some(child));
  447. Ok(ServerReadyData {
  448. url,
  449. username,
  450. password,
  451. is_sidecar: true,
  452. })
  453. }
  454. .map(move |res| {
  455. let _ = server_ready_tx.send(res);
  456. }),
  457. )
  458. }
  459. ServerConnection::Existing { url } => {
  460. let _ = server_ready_tx.send(Ok(ServerReadyData {
  461. url: url.to_string(),
  462. username: None,
  463. password: None,
  464. is_sidecar: false,
  465. }));
  466. None
  467. }
  468. };
  469. tracing::info!("server connection started");
  470. if let Some(cli_health_check) = cli_health_check {
  471. if let Some(sqlite_done_rx) = sqlite_done {
  472. let _ = sqlite_done_rx.await;
  473. }
  474. tokio::spawn(cli_health_check);
  475. }
  476. let _ = server_ready_rx.await;
  477. tracing::info!("Loading task finished");
  478. }
  479. })
  480. .map_err(|_| ())
  481. .shared();
  482. let loading_window = if needs_sqlite_migration
  483. && timeout(Duration::from_secs(1), loading_task.clone())
  484. .await
  485. .is_err()
  486. {
  487. tracing::debug!("Loading task timed out, showing loading window");
  488. let loading_window = LoadingWindow::create(&app).expect("Failed to create loading window");
  489. sleep(Duration::from_secs(1)).await;
  490. Some(loading_window)
  491. } else {
  492. tracing::debug!("Showing main window without loading window");
  493. MainWindow::create(&app).expect("Failed to create main window");
  494. None
  495. };
  496. let _ = loading_task.await;
  497. tracing::info!("Loading done, completing initialisation");
  498. let _ = init_tx.send(InitStep::Done);
  499. if loading_window.is_some() {
  500. loading_window_complete.await;
  501. tracing::info!("Loading window completed");
  502. }
  503. MainWindow::create(&app).expect("Failed to create main window");
  504. if let Some(loading_window) = loading_window {
  505. let _ = loading_window.close();
  506. }
  507. }
  508. fn setup_app(app: &tauri::AppHandle, init_rx: watch::Receiver<InitStep>) {
  509. #[cfg(any(target_os = "linux", all(debug_assertions, windows)))]
  510. app.deep_link().register_all().ok();
  511. app.manage(InitState { current: init_rx });
  512. }
  513. fn spawn_cli_sync_task(app: AppHandle) {
  514. tokio::spawn(async move {
  515. if let Err(e) = sync_cli(app) {
  516. tracing::error!("Failed to sync CLI: {e}");
  517. }
  518. });
  519. }
  520. enum ServerConnection {
  521. Existing {
  522. url: String,
  523. },
  524. CLI {
  525. url: String,
  526. username: Option<String>,
  527. password: Option<String>,
  528. child: CommandChild,
  529. health_check: server::HealthCheck,
  530. },
  531. }
  532. async fn setup_server_connection(app: AppHandle) -> ServerConnection {
  533. let custom_url = get_saved_server_url(&app).await;
  534. tracing::info!(?custom_url, "Attempting server connection");
  535. if let Some(url) = &custom_url
  536. && server::check_health_or_ask_retry(&app, url).await
  537. {
  538. tracing::info!(%url, "Connected to custom server");
  539. // If the default server is already local, no need to also spawn a sidecar
  540. if server::is_localhost_url(url) {
  541. return ServerConnection::Existing { url: url.clone() };
  542. }
  543. // Remote default server: fall through and also spawn a local sidecar
  544. }
  545. let local_port = get_sidecar_port();
  546. let hostname = "127.0.0.1";
  547. let local_url = format!("http://{hostname}:{local_port}");
  548. tracing::debug!(url = %local_url, "Checking health of local server");
  549. if server::check_health(&local_url, None).await {
  550. tracing::info!(url = %local_url, "Health check OK, using existing server");
  551. return ServerConnection::Existing { url: local_url };
  552. }
  553. let password = uuid::Uuid::new_v4().to_string();
  554. tracing::info!("Spawning new local server");
  555. let (child, health_check) =
  556. server::spawn_local_server(app, hostname.to_string(), local_port, password.clone());
  557. ServerConnection::CLI {
  558. url: local_url,
  559. username: Some("opencode".to_string()),
  560. password: Some(password),
  561. child,
  562. health_check,
  563. }
  564. }
  565. fn get_sidecar_port() -> u32 {
  566. option_env!("OPENCODE_PORT")
  567. .map(|s| s.to_string())
  568. .or_else(|| std::env::var("OPENCODE_PORT").ok())
  569. .and_then(|port_str| port_str.parse().ok())
  570. .unwrap_or_else(|| {
  571. TcpListener::bind("127.0.0.1:0")
  572. .expect("Failed to bind to find free port")
  573. .local_addr()
  574. .expect("Failed to get local address")
  575. .port()
  576. }) as u32
  577. }
  578. fn sqlite_file_exists() -> bool {
  579. let Ok(path) = opencode_db_path() else {
  580. return true;
  581. };
  582. path.exists()
  583. }
  584. fn opencode_db_path() -> Result<PathBuf, &'static str> {
  585. let xdg_data_home = env::var_os("XDG_DATA_HOME").filter(|v| !v.is_empty());
  586. let data_home = match xdg_data_home {
  587. Some(v) => PathBuf::from(v),
  588. None => {
  589. let home = dirs::home_dir().ok_or("cannot determine home directory")?;
  590. home.join(".local").join("share")
  591. }
  592. };
  593. Ok(data_home.join("opencode").join("opencode.db"))
  594. }
  595. // Creates a `once` listener for the specified event and returns a future that resolves
  596. // when the listener is fired.
  597. // Since the future creation and awaiting can be done separately, it's possible to create the listener
  598. // synchronously before doing something, then awaiting afterwards.
  599. fn event_once_fut<T: tauri_specta::Event + serde::de::DeserializeOwned>(
  600. app: &AppHandle,
  601. ) -> impl Future<Output = ()> {
  602. let (tx, rx) = oneshot::channel();
  603. T::once(app, |_| {
  604. let _ = tx.send(());
  605. });
  606. async {
  607. let _ = rx.await;
  608. }
  609. }