nseqe 0.1.0

A Rust library for managing and processing sequences of events.
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

use std::{collections::HashMap, fmt::Debug, net::SocketAddr, sync::Arc};

use tokio::{
    net::TcpStream,
    sync::{Mutex, Notify},
    time::Instant,
};
use tracing::{event, instrument};

use crate::action::Action;

/// Represents an event that occurs when data is received over a network socket.
#[derive(Debug, Clone)]
pub struct ReceiveEvent {
    pub instant: Instant,
    pub from: SocketAddr,
    pub to: SocketAddr,
    pub buffer: Vec<u8>,
}

/// Represents an event that occurs when data is sent over a network socket.
#[derive(Debug, Clone)]
pub struct SendEvent {
    pub instant: Instant,
    pub from: SocketAddr,
    pub to: SocketAddr,
    pub buffer: Vec<u8>,
}

/// Represents an event that occurs when a connection is established between two sockets.
#[derive(Debug, Clone)]
pub struct ConnectEvent {
    pub instant: Instant,
    pub from: SocketAddr,
    pub to: SocketAddr,
}

/// Represents the context of a node, containing its TCP streams and events.
#[derive(Debug)]
pub struct NodeContext {
    pub tcp_streams: HashMap<SocketAddr, TcpStream>,
    pub receive_events: Vec<ReceiveEvent>,
    pub receive_notifier: Arc<Notify>,
    pub send_events: Vec<SendEvent>,
    pub connect_events: Vec<ConnectEvent>,
    pub connect_notifier: Arc<Notify>,
}

pub type Ctx = Arc<Mutex<NodeContext>>;

/// Represents a node in the network, which can perform actions and maintain its context.
pub struct Node {
    name: String,
    actions: Vec<Box<dyn Action>>,
    ctx: Ctx,
}

impl Debug for Node {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "node-{:?}", self.name)
    }
}

impl Node {
    pub fn new(name: &str) -> Self {
        Node {
            name: name.to_string(),
            actions: Vec::new(),
            ctx: Arc::new(Mutex::new(NodeContext {
                tcp_streams: HashMap::new(),
                receive_events: Vec::new(),
                receive_notifier: Arc::new(Notify::new()),
                send_events: Vec::new(),
                connect_events: Vec::new(),
                connect_notifier: Arc::new(Notify::new()),
            })),
        }
    }

    pub fn name(&self) -> &str {
        &self.name
    }

    pub fn add_action<T>(&mut self, action: T)
    where
        T: Action + 'static,
    {
        self.actions.push(Box::new(action));
    }

    pub fn ctx(&self) -> Ctx {
        Arc::clone(&self.ctx)
    }

    /// Starts the node, executing all its actions in order.
    #[instrument(name = "node_start", level = "info", skip(self), fields(node = %self.name))]
    pub async fn start(&mut self) {
        for action in self.actions.drain(..) {
            #[instrument(name = "action", level = "info", skip(ctx), fields(action = %action.name()))]
            async fn run_action(action: &dyn Action, ctx: Ctx) {
                if let Err(e) = action.perform(ctx).await {
                    event!(tracing::Level::ERROR, "Error performing action: {:?}", e);
                }
            }

            run_action(&*action, self.ctx.clone()).await;
        }

        event!(tracing::Level::INFO, "All actions performed");
        event!(tracing::Level::INFO, "Node {} finished", self.name);
        event!(
            tracing::Level::INFO,
            "Received messages: {:?}",
            self.ctx.lock().await.receive_events
        );
    }
}