3 Commits

Author SHA1 Message Date
564e606ac7 Properly handle start sync thread issue 2025-11-19 17:15:54 +01:00
7b691962a0 Can get sync thread status 2025-11-19 16:34:00 +01:00
1e00d24a8b Can request sync thread stop 2025-11-19 15:51:15 +01:00
4 changed files with 87 additions and 5 deletions

View File

@@ -20,3 +20,40 @@ pub async fn start_sync(
}
}
}
/// Stop sync thread
pub async fn stop_sync(
client: MatrixClientExtractor,
manager: web::Data<ActorRef<MatrixManagerMsg>>,
) -> HttpResult {
match ractor::cast!(
manager,
MatrixManagerMsg::StopSyncThread(client.auth.user.email.clone())
) {
Ok(_) => Ok(HttpResponse::Accepted().finish()),
Err(e) => {
log::error!("Failed to stop sync thread: {e}");
Ok(HttpResponse::InternalServerError().finish())
}
}
}
#[derive(serde::Serialize)]
struct GetSyncStatusResponse {
started: bool,
}
/// Get sync thread status
pub async fn status(
client: MatrixClientExtractor,
manager: web::Data<ActorRef<MatrixManagerMsg>>,
) -> HttpResult {
let started = ractor::call!(
manager.as_ref(),
MatrixManagerMsg::SyncThreadGetStatus,
client.auth.user.email
)
.expect("RPC to Matrix Manager failed");
Ok(HttpResponse::Ok().json(GetSyncStatusResponse { started }))
}

View File

@@ -39,9 +39,13 @@ impl FromRequest for MatrixClientExtractor {
matrix_manager_actor,
MatrixManagerMsg::GetClient,
auth.user.email.clone()
)
.expect("Failed to query manager actor!")
.expect("Failed to get client!");
);
let client = match client {
Ok(Ok(client)) => client,
Ok(Err(err)) => panic!("Failed to get client! {err:?}"),
Err(err) => panic!("Failed to query manager actor! {err:#?}"),
};
Ok(Self { auth, client })
})

View File

@@ -125,6 +125,14 @@ async fn main() -> std::io::Result<()> {
"/api/matrix_sync/start",
web::post().to(matrix_sync_thread_controller::start_sync),
)
.route(
"/api/matrix_sync/stop",
web::post().to(matrix_sync_thread_controller::stop_sync),
)
.route(
"/api/matrix_sync/status",
web::get().to(matrix_sync_thread_controller::status),
)
})
.workers(4)
.bind(&AppConfig::get().listen_address)?

View File

@@ -15,6 +15,8 @@ pub enum MatrixManagerMsg {
GetClient(UserEmail, RpcReplyPort<anyhow::Result<MatrixClient>>),
DisconnectClient(UserEmail),
StartSyncThread(UserEmail),
StopSyncThread(UserEmail),
SyncThreadGetStatus(UserEmail, RpcReplyPort<bool>),
SyncThreadTerminated(UserEmail, MatrixSyncTaskID),
}
@@ -37,6 +39,15 @@ impl Actor for MatrixManagerActor {
})
}
async fn post_stop(
&self,
_myself: ActorRef<Self::Msg>,
_state: &mut Self::State,
) -> Result<(), ActorProcessingErr> {
log::error!("[!] [!] Matrix Manager Actor stopped!");
Ok(())
}
async fn handle(
&self,
myself: ActorRef<Self::Msg>,
@@ -113,10 +124,32 @@ impl Actor for MatrixManagerActor {
// Start thread
log::debug!("Starting sync thread for {email:?}");
let thread_id =
start_sync_thread(client.clone(), state.broadcast_sender.clone(), myself)
.await?;
match start_sync_thread(client.clone(), state.broadcast_sender.clone(), myself)
.await
{
Ok(thread_id) => thread_id,
Err(e) => {
log::error!("Failed to start sync thread! {e}");
return Ok(());
}
};
state.running_sync_threads.insert(email, thread_id);
}
MatrixManagerMsg::StopSyncThread(email) => {
if let Some(thread_id) = state.running_sync_threads.get(&email)
&& let Err(e) = state
.broadcast_sender
.send(BroadcastMessage::StopSyncThread(thread_id.clone()))
{
log::error!("Failed to request sync thread stop: {e}");
}
}
MatrixManagerMsg::SyncThreadGetStatus(email, reply) => {
let started = state.running_sync_threads.contains_key(&email);
if let Err(e) = reply.send(started) {
log::error!("Failed to send sync thread status! {e}");
}
}
MatrixManagerMsg::SyncThreadTerminated(email, task_id) => {
if state.running_sync_threads.get(&email) == Some(&task_id) {
log::info!(