-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Rename OpenWorkspaceByPathError to OpenWorkspaceError Use Option<T> functions instead of doing it manually * Initial implementation of the ApiManager Move LibSqlStore and InMemoryEvents to fpx * Add fpx.toml * Resolve warnings * Fix tests * Add simple logging with Tracing Rename CI from build to build_cli * Add legacy and fpx api manager Fix issue with duplicate recent workspace entries * Ensure that the lock stays locked long enough (Rust API) Previously there was a possibility that start would be called twice and two processes would be started, but only one PID would be stored Add more docs, logs, etc * Remove unsused file * Fix build issue Use 8788 as api port * remove todo's, they are captured by issues
- Loading branch information
Showing
33 changed files
with
357 additions
and
103 deletions.
There are no files selected for viewing
File renamed without changes.
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| #[cfg(feature = "fpx-api")] | ||
| mod fpx_api; | ||
|
|
||
| #[cfg(feature = "fpx-api")] | ||
| pub use fpx_api::*; | ||
|
|
||
| #[cfg(not(feature = "fpx-api"))] | ||
| mod legacy; | ||
|
|
||
| #[cfg(not(feature = "fpx-api"))] | ||
| pub use legacy::*; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| use fpx::api; | ||
| use fpx::config::FpxConfig; | ||
| use fpx::data::libsql_store::LibsqlStore; | ||
| use fpx::events::memory::InMemoryEvents; | ||
| use fpx::service::Service; | ||
| use std::sync::{Arc, Mutex}; | ||
| use tauri::async_runtime::spawn; | ||
| use tokio::sync::broadcast::error::RecvError; | ||
| use tracing::{error, info, trace, warn}; | ||
|
|
||
| #[derive(Debug, Default)] | ||
| pub struct ApiManager { | ||
| // Sending a message on this channel will shutdown the axum server. | ||
| shutdown_tx: Mutex<Option<tokio::sync::oneshot::Sender<()>>>, | ||
| } | ||
|
|
||
| impl ApiManager { | ||
| pub fn start_api(&self, fpx_config: FpxConfig) { | ||
| let mut shutdown_tx = self.shutdown_tx.lock().expect("lock is poisoned"); | ||
| if let Some(shutdown_tx) = shutdown_tx.take() { | ||
| // shutdown any existing api server | ||
| let _ = shutdown_tx.send(()); | ||
| } | ||
|
|
||
| // Start a listener early, so that we can handle the issue where another | ||
| // process is listening already on that specific port. | ||
| let listen_port = fpx_config.listen_port.unwrap_or(6767); | ||
| let listener = std::net::TcpListener::bind(format!("127.0.0.1:{listen_port}")).unwrap(); | ||
| listener.set_nonblocking(true).unwrap(); | ||
|
|
||
| let (shutdown, on_shutdown) = tokio::sync::oneshot::channel::<()>(); | ||
| *shutdown_tx = Some(shutdown); | ||
|
|
||
| spawn(async move { | ||
| let store = LibsqlStore::in_memory().await.unwrap(); | ||
| LibsqlStore::migrate(&store).await.unwrap(); | ||
| let store = Arc::new(store); | ||
|
|
||
| // Create a event sink that simply logs | ||
| let events = InMemoryEvents::new(); | ||
| let events = Arc::new(events); | ||
|
|
||
| // Our current implementation simply logs the events. | ||
| let mut reader = events.subscribe(); | ||
| spawn(async move { | ||
| loop { | ||
| match reader.recv().await { | ||
| Ok(message) => { | ||
| // Here we can do something with events, like | ||
| // emitting them to the frontend: | ||
| // window.emit("api_message", message).expect("emit failed"); | ||
| info!("Received message: {:?}", message); | ||
| } | ||
| Err(RecvError::Lagged(i)) => { | ||
| warn!(lagged = i, "Event reader lagged behind"); | ||
| } | ||
| Err(RecvError::Closed) => { | ||
| trace!("Event reader loop stopped"); | ||
| break; | ||
| } | ||
| } | ||
| } | ||
| }); | ||
|
|
||
| let service = Service::new(store.clone(), events.clone()); | ||
|
|
||
| let app = api::Builder::new() | ||
| .enable_compression() | ||
| .build(service.clone(), store.clone()); | ||
|
|
||
| let listener = tokio::net::TcpListener::from_std(listener).unwrap(); | ||
| let api_server = axum::serve(listener, app).with_graceful_shutdown(async { | ||
| // Once we receive something on the [`on_shutdown`] channel, | ||
| // we'll resolve this future, and thus axum will shutdown. | ||
| // We are wrapping this in another future because of the | ||
| // incompatible return type of the oneshot channel. | ||
| let _ = on_shutdown.await; | ||
| trace!("Received API shutdown signal"); | ||
| }); | ||
|
|
||
| if let Err(err) = api_server.await { | ||
| error!(?err, "API server returned an error"); | ||
| }; | ||
| }); | ||
| } | ||
|
|
||
| pub fn stop_api(&self) { | ||
| let mut shutdown_tx = self.shutdown_tx.lock().expect("lock is poisoned"); | ||
| if let Some(shutdown_tx) = shutdown_tx.take() { | ||
| // shutdown any existing api servers | ||
| let _ = shutdown_tx.send(()); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| use fpx::config::FpxConfig; | ||
| use nix::sys::signal::{killpg, Signal}; | ||
| use nix::unistd::Pid; | ||
| use std::os::unix::process::CommandExt; | ||
| use std::process; | ||
| use std::sync::Mutex; | ||
| use tauri::async_runtime::spawn; | ||
| use tracing::{error, trace, warn}; | ||
|
|
||
| #[derive(Debug, Default)] | ||
| pub struct ApiManager { | ||
| api_pid: Mutex<Option<Pid>>, | ||
| } | ||
|
|
||
| impl ApiManager { | ||
| /// Start a API server. If a API pid is already set, then that will first be | ||
| /// shutdown. | ||
| pub fn start_api(&self, fpx_config: FpxConfig) { | ||
| // Get a lock for the duration of this function | ||
| let mut api_pid = self.api_pid.lock().expect("lock is poisoned"); | ||
|
|
||
| // If there is a API pid already there, then first send the SIGTERM | ||
| // signal to that process group. | ||
| if let Some(api_pid) = api_pid.take() { | ||
| // shutdown any existing api server | ||
| Self::send_sigterm_signal(api_pid); | ||
| } | ||
|
|
||
| // Create some environment variables overrides based on the fpx.toml | ||
| let mut envs: Vec<(&str, String)> = vec![]; | ||
| if let Some(listen_port) = fpx_config.listen_port { | ||
| envs.push(("FPX_PORT", listen_port.to_string())); | ||
| } | ||
|
|
||
| // Start the process using pnpm. The process_group=0 will ensure that | ||
| // the process group ID is the same as the root process ID. | ||
| let mut child_process = process::Command::new("pnpm") | ||
| .arg("dev:api") | ||
| .process_group(0) // | ||
| .envs(envs) | ||
| .spawn() | ||
| .expect("failed to execute pnpm dev:api"); | ||
|
|
||
| // Once the process is running, get the pid and store it in the mutex, | ||
| // so that we can send signals to it later. | ||
| let pid = child_process.id(); | ||
| *api_pid = Some(Pid::from_raw(pid as i32)); | ||
|
|
||
| // Spawn a task to wait for the child process to exit, and potentially | ||
| // log an error | ||
| spawn(async move { | ||
| let result = child_process.wait(); | ||
| if let Err(err) = result { | ||
| error!(?err, api_pid=?pid, "child process exited with error"); | ||
| } else { | ||
| trace!(api_pid=?pid, "API server exited successfully"); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| /// Sends the SIGTERM signal to the API process group. If no API pid was set | ||
| /// then this function will do nothing. | ||
| pub fn stop_api(&self) { | ||
| let Some(api_pid) = self.api_pid.lock().expect("lock is poisoned").take() else { | ||
| trace!("No API running"); | ||
| return; | ||
| }; | ||
|
|
||
| Self::send_sigterm_signal(api_pid) | ||
| } | ||
|
|
||
| /// Send the SIGTERM signal to the specified process group. | ||
| /// | ||
| /// This uses a Process ID type instead of a specific process group ID as | ||
| /// that does not exist. | ||
| fn send_sigterm_signal(api_pid: Pid) { | ||
| trace!(?api_pid, "sending SIGTERM signal to API process group"); | ||
|
|
||
| let result = killpg(api_pid, Signal::SIGTERM); | ||
| if let Err(errno) = result { | ||
| warn!( | ||
| ?errno, | ||
| ?api_pid, | ||
| "failed to send SIGNTERM signal to API process group" | ||
| ); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.