bpm_engine_runtime/
user_task_completed_handler.rs

1//! UserTaskCompletedHandler: UserTaskCompleted -> move token, emit TokenArrived.
2
3use async_trait::async_trait;
4use bpm_engine_core::{payloads, EngineEvent};
5use tracing::info;
6
7use super::handler::{EngineContext, EventHandler};
8use super::transition::move_token;
9
10pub struct UserTaskCompletedHandler;
11
12#[async_trait]
13impl EventHandler for UserTaskCompletedHandler {
14    async fn handle(&self, event: &EngineEvent, ctx: &mut EngineContext) -> Vec<EngineEvent> {
15        let EngineEvent::UserTaskCompleted(e) = event else {
16            return vec![];
17        };
18        info!(instance_id = %e.instance_id, node_id = %e.node_id, "user task completed");
19        let Some(process_store) = ctx.process_store.as_ref() else {
20            return vec![];
21        };
22        let Some(process_def_store) = ctx.process_def_store.as_ref() else {
23            return vec![];
24        };
25        let Ok(Some(mut instance)) = process_store.load(&e.instance_id).await else {
26            return vec![];
27        };
28        let Ok(Some(def)) = process_def_store.load(&instance.process_def_id).await else {
29            return vec![];
30        };
31        let Some(node) = def.nodes.get(e.node_id.as_str()) else {
32            return vec![];
33        };
34        instance.tokens.retain(|t| t.node_id != e.node_id);
35        let new_tokens = move_token(node);
36        let mut out = vec![];
37        for t in &new_tokens {
38            out.push(EngineEvent::TokenArrived(payloads::TokenArrived {
39                instance_id: e.instance_id.clone(),
40                token_id: t.id.clone(),
41                node_id: t.node_id.clone(),
42            }));
43        }
44        instance.tokens.extend(new_tokens);
45        let _ = process_store.save(&instance).await;
46        out
47    }
48}
bpm_engine_runtime/user_task_completed_handler.rs

bpm_engine_runtime/
user_task_completed_handler.rs
