kproc 0.7.0

Knowledge Processing library.
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

//! Cache values, used to avoid a stream to clog

pub use kproc_pmacros::Processor;
use std::collections::LinkedList;

use crate::processor::{InputStreams, OutputStreams};
use crate::{processor, Result};

/// Inputs to cache
#[derive(Clone, InputStreams)]
pub struct Inputs<T>
where
  T: Clone,
{
  t: T,
}

/// Outputs to cache
#[derive(Clone, OutputStreams)]
pub struct Outputs<T>
where
  T: Clone,
{
  t: T,
}

/// Allow to cache a value in a kproc graph.
#[derive(Processor)]
#[streams(Inputs<T>, Outputs<T>)]
pub struct Cache<T>
where
  T: Clone,
{
  _t: std::marker::PhantomData<T>,
}

impl<T> processor::Processor for Cache<T>
where
  T: Clone,
{
  // async fn process_one(&self, inputs: Self::Inputs) -> Self::Outputs {}
  async fn start(self, mut inputs: Self::InputStreams, mut outputs: Self::OutputStreams)
  {
    let mut cache = LinkedList::<T>::new();
    loop
    {
      let i = inputs.next().await.unwrap();
      if outputs.is_full()
      {
        cache.push_back(i.t);
      }
      else
      {
        while !cache.is_empty() && !outputs.is_full()
        {
          let _ = outputs
            .next(Outputs {
              t: cache.pop_front().unwrap(),
            })
            .await;
        }
        if outputs.is_full()
        {
          cache.push_back(i.t);
        }
        else
        {
          let _ = outputs.next(Outputs { t: i.t }).await;
        }
      }
    }
  }
}