nitinol-process 0.1.0

Pseudo-actor type process library for Nitinol using green thread of tokio
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

use nitinol_core::identifier::ToEntityId;
use crate::channel::ProcessApplier;
use crate::{Process, Context};
use crate::errors::AlreadyExist;
use crate::receptor::Receptor;
use crate::registry::ProcessRegistry;

pub async fn run<T: Process>(
    id: impl ToEntityId,
    entity: T,
    start_seq: i64,
    registry: ProcessRegistry
) -> Result<Receptor<T>, AlreadyExist> {
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Box<dyn ProcessApplier<T>>>();

    let entity_id = id.to_entity_id();
    let refs = Receptor { channel: tx };
    
    let context = Context::new(start_seq, registry.clone());
    
    registry.register(entity_id.clone(), refs.clone()).await?;
    
    tokio::spawn(async move {
        let id = entity_id;
        let mut entity = entity;
        let mut context = context;
        let registry = registry;
        
        entity.start(&mut context).await;
        
        while let Some(rx) = rx.recv().await {
            if let Err(e) = rx.apply(&mut entity, &mut context).await {
                tracing::error!("{e}");
            }
            
            if !context.is_active().await {
                tracing::warn!("lifecycle ended.");
                break;
            }
        }
        
        if let Err(e) = registry.deregister(&id).await {
            tracing::error!("{e}");
        }
        
        entity.stop(&mut context).await;
    });
    
    Ok(refs)
}