dagrs 0.8.1

Dagrs follows the concept of Flow-based Programming and is suitable for the execution of multiple tasks with graph-like dependencies. Dagrs has the characteristics of high performance and asynchronous execution. It provides users with a convenient programming interface.
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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

use crate::connection::{in_channel::InChannels, out_channel::OutChannels};
use crate::node::{Node, NodeId, NodeName, NodeTable};
use crate::utils::{env::EnvVar, output::FlowControl, output::Output};
use async_trait::async_trait;
use std::sync::Arc;

/// Router trait for routing logic.
///
/// This trait allows users to implement dynamic branching in the graph.
/// A `Router` decides which downstream path(s) execution should follow.
#[async_trait]
pub trait Router: Send + Sync {
    /// Determine the next nodes to execute.
    ///
    /// # Arguments
    /// * `input` - Input channels to read data from upstream nodes.
    /// * `output` - Output channels to send data to downstream nodes.
    /// * `env` - The execution environment variables.
    ///
    /// # Returns
    /// A vector of `usize` representing the `NodeId`s of the nodes that should be activated.
    /// Any downstream nodes connected to this router that are NOT in this list will be skipped (pruned).
    ///
    /// # Data Transmission vs. Route Selection
    /// The `route` method serves two purposes:
    /// 1. **Data Transmission**: It can optionally send data to downstream nodes via `output`.
    ///    This is useful if the router acts as a dispatcher or load balancer.
    /// 2. **Route Selection**: It MUST return the IDs of the nodes that should run.
    ///    Even if data is sent to a node, if its ID is not in the returned vector, it will NOT run.
    async fn route(
        &self,
        input: &mut InChannels,
        output: &OutChannels,
        env: Arc<EnvVar>,
    ) -> Vec<usize>;

    /// Reset the router state.
    /// This is called when the graph is reset.
    fn reset(&mut self) {}
}

/// Router node implementation.
///
/// Wraps a user-provided `Router` implementation and integrates it into the graph execution.
/// When executed, it delegates to `Router::route` and converts the result into a `FlowControl::Branch` instruction.
pub struct RouterNode {
    id: NodeId,
    name: NodeName,
    in_channels: InChannels,
    out_channels: OutChannels,
    router: Box<dyn Router>,
}

impl RouterNode {
    /// Create a new RouterNode
    pub fn new(name: NodeName, router: impl Router + 'static, node_table: &mut NodeTable) -> Self {
        Self {
            id: node_table.alloc_id_for(&name),
            name,
            in_channels: InChannels::default(),
            out_channels: OutChannels::default(),
            router: Box::new(router),
        }
    }
}

#[async_trait]
impl Node for RouterNode {
    fn id(&self) -> NodeId {
        self.id
    }
    fn name(&self) -> NodeName {
        self.name.clone()
    }
    fn input_channels(&mut self) -> &mut InChannels {
        &mut self.in_channels
    }
    fn output_channels(&mut self) -> &mut OutChannels {
        &mut self.out_channels
    }
    async fn run(&mut self, env: Arc<EnvVar>) -> Output {
        let routes = self
            .router
            .route(&mut self.in_channels, &self.out_channels, env)
            .await;
        Output::Flow(FlowControl::Branch(routes))
    }
    fn is_condition(&self) -> bool {
        true
    }

    fn reset(&mut self) {
        self.router.reset();
    }
}