remotia_core/processors/
pool_switch.rs

1use std::{collections::HashMap, fmt::Debug, hash::Hash};
2
3use async_trait::async_trait;
4use log::debug;
5use rand::prelude::{SliceRandom, ThreadRng};
6
7use crate::{
8    pipeline::{feeder::PipelineFeeder, Pipeline},
9    traits::{FrameProcessor, FrameProperties},
10};
11
12pub struct PoolingSwitch<F, P, K> {
13    property_key: P,
14    entries: Vec<(K, PipelineFeeder<F>)>
15}
16
17impl<F, P, K> PoolingSwitch<F, P, K> where
18    F: Debug + Default + Send
19{
20    pub fn new(property_key: P) -> Self {
21        Self {
22            property_key,
23            entries: Vec::new(),
24        }
25    }
26
27    pub fn entry(mut self, key: K, pipeline: &mut Pipeline<F>) -> Self where
28        F: 'static
29    {
30        self.entries.push((key, pipeline.get_feeder()));
31        self
32    }
33}
34
35#[async_trait]
36impl<F, P, K> FrameProcessor<F> for PoolingSwitch<F, P, K> where
37    P: Copy + Send,
38    K: Copy + Send,
39    F: Debug + FrameProperties<P, K> + Send + 'static
40{
41    async fn process(&mut self, mut frame_data: F) -> Option<F> {
42        let (key, feeder) = self.entries.choose(&mut rand::thread_rng()).unwrap();
43
44        frame_data.set(self.property_key, *key);
45        feeder.feed(frame_data);
46
47        None
48    }
49}
50
51pub struct DepoolingSwitch<F, P, K> {
52    property_key: P,
53    entries: HashMap<K, PipelineFeeder<F>>
54}
55
56impl<F, P, K> DepoolingSwitch<F, P, K> {
57    pub fn new(property_key: P) -> Self {
58        Self {
59            property_key,
60            entries: HashMap::new(),
61        }
62    }
63
64    pub fn entry(mut self, key: K, pipeline: &mut Pipeline<F>) -> Self where
65        K: Hash + Eq,
66        F: Debug + Default + Send + 'static
67    {
68        self.entries.insert(key, pipeline.get_feeder());
69        self
70    }
71}
72
73#[async_trait]
74impl<F, P, K> FrameProcessor<F> for DepoolingSwitch<F, P, K> 
75where
76    P: Send,
77    K: Eq + Hash + Send,
78    F: Debug + FrameProperties<P, K> + Send + 'static
79{
80    async fn process(&mut self, frame_data: F) -> Option<F> {
81        let key = frame_data.get(&self.property_key).unwrap();
82        let feeder = self.entries.get(&key).unwrap();
83
84        feeder.feed(frame_data);
85
86        None
87    }
88}
89
remotia_core/processors/pool_switch.rs

remotia_core/processors/
pool_switch.rs
