remotia-core 0.1.2

Core components of remotia, an open source framework for the development of remote rendering software in pure Rust
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
118

use std::fmt::Debug;

use log::info;
use tokio::{sync::mpsc::{self, UnboundedSender}, task::JoinHandle};

use self::{component::Component, feeder::PipelineFeeder};

pub mod component;
pub mod feeder;
pub mod registry;

pub struct Pipeline<F> {
    components: Vec<Component<F>>,
    feeding_sender: Option<UnboundedSender<F>>,

    tag: String,

    bound: bool,

    to_be_feedable: bool,
}

impl<F: Debug + Default + Send + 'static> Pipeline<F> {
    pub fn new() -> Self {
        Self {
            components: Vec::new(),
            feeding_sender: None,

            tag: "".to_string(),

            bound: false,

            to_be_feedable: false,
        }
    }

    pub fn singleton(component: Component<F>) -> Self {
        Self::new().link(component)
    }

    pub fn link(mut self, component: Component<F>) -> Self {
        self.components.push(component);
        self
    }

    pub fn get_feeder(&mut self) -> PipelineFeeder<F> {
        if self.to_be_feedable {
            self.make_feedable();
        }

        let sender = self.feeding_sender.as_ref().unwrap().clone();
        PipelineFeeder::new(sender)
    }

    pub fn run(mut self) -> Vec<JoinHandle<()>> {
        info!("[{}] Launching threads...", self.tag);

        if !self.bound {
            self.bind();
        }

        if self.to_be_feedable {
            self.make_feedable();
        }

        let mut handles = Vec::new();

        for component in self.components {
            let handle = component.launch();
            handles.push(handle);
        }

        handles
    }

    fn bind(&mut self) {
        info!("[{}] Binding channels...", self.tag);

        for i in 0..self.components.len()-1 {
            let (sender, receiver) = mpsc::unbounded_channel::<F>();

            let src_component = self.components.get_mut(i).unwrap();
            src_component.set_sender(sender);

            let dst_component = self.components.get_mut(i + 1).unwrap();
            dst_component.set_receiver(receiver);
        }

        self.bound = true;
    }

    fn make_feedable(&mut self) {
        let head = self.components.get_mut(0).unwrap();

        let (sender, receiver) = mpsc::unbounded_channel::<F>();
        self.feeding_sender = Some(sender);

        head.set_receiver(receiver);

        self.to_be_feedable = false;
    }

    pub fn tag(mut self, tag: &str) -> Self {
        self.tag = tag.to_string();
        self
    }

    pub fn feedable(mut self) -> Self {
        self.to_be_feedable = true;
        self
    }
}

impl<F: Default + Debug + Send + 'static> Default for Pipeline<F> {
    fn default() -> Self {
        Self::new()
    }
}