wingfoil 3.0.1

graph based stream processing framework
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132

//! Adpaters to convert an Iterators into a Streams

use crate::queue::ValueAt;
use crate::types::*;
use anyhow::anyhow;

use std::cmp::Ordering;

type Peeker<T> = Box<std::iter::Peekable<Box<dyn Iterator<Item = ValueAt<T>>>>>;

/// Wraps an Iterator and exposes it as a stream of Vectors
pub struct IteratorStream<T: Element> {
    peekable: Peeker<T>,
    value: Vec<T>,
}

fn add_callback<T>(peekable: &mut Peeker<T>, state: &mut GraphState) -> anyhow::Result<bool> {
    match peekable.peek() {
        Some(value_at) => {
            state.add_callback(value_at.time);
            Ok(true)
        }
        None => Ok(false),
    }
}

impl<T: Element> MutableNode for IteratorStream<T> {
    fn cycle(&mut self, state: &mut GraphState) -> anyhow::Result<bool> {
        self.value.clear();
        {
            while let Some(value_at) = self.peekable.peek() {
                if value_at.time == state.time() {
                    let val = self.peekable.next().unwrap().value.clone();
                    self.value.push(val);
                } else {
                    break;
                }
            }
        }
        add_callback(&mut self.peekable, state)
    }

    fn upstreams(&self) -> UpStreams {
        UpStreams::default()
    }

    fn start(&mut self, state: &mut GraphState) -> anyhow::Result<()> {
        add_callback(&mut self.peekable, state)?;
        Ok(())
    }
}

impl<T: Element> StreamPeekRef<Vec<T>> for IteratorStream<T> {
    fn peek_ref(&self) -> &Vec<T> {
        &self.value
    }
}

impl<T> IteratorStream<T>
where
    T: Element + 'static,
{
    pub fn new(it: Box<dyn Iterator<Item = ValueAt<T>>>) -> Self {
        let peekable = Box::new(it.peekable());
        Self {
            peekable,
            value: vec![],
        }
    }
}

/// Wraps an Iterator and exposes it as a [Stream] of values.
/// The source must be strictly ascending in time.   If the source
/// can tick multiple times at one time, then you can use an
/// [IteratorStream] instead, which emits `Vec<T>`.
pub struct SimpleIteratorStream<T: Element> {
    peekable: Peeker<T>,
    value: T,
}

impl<T: Element> MutableNode for SimpleIteratorStream<T> {
    fn cycle(&mut self, state: &mut GraphState) -> anyhow::Result<bool> {
        {
            let val_at1 = self.peekable.next().unwrap();
            self.value = val_at1.value;

            let optional_val_at2 = self.peekable.peek();
            if let Some(val_at2) = optional_val_at2 {
                match val_at1.time.cmp(&val_at2.time) {
                    Ordering::Greater => {
                        return Err(anyhow!("source time was descending!"));
                    }
                    Ordering::Equal => {
                        return Err(anyhow!(
                            "source produced multiple ticks for same time, use IteratorStream instead"
                        ));
                    }
                    Ordering::Less => {}
                }
            }
        }
        add_callback(&mut self.peekable, state)
    }

    fn upstreams(&self) -> UpStreams {
        UpStreams::default()
    }

    fn start(&mut self, state: &mut GraphState) -> anyhow::Result<()> {
        add_callback(&mut self.peekable, state)?;
        Ok(())
    }
}

impl<T: Element> StreamPeekRef<T> for SimpleIteratorStream<T> {
    fn peek_ref(&self) -> &T {
        &self.value
    }
}

impl<T> SimpleIteratorStream<T>
where
    T: Element + 'static,
{
    pub fn new(it: Box<dyn Iterator<Item = ValueAt<T>>>) -> SimpleIteratorStream<T> {
        let peekable = Box::new(it.peekable());
        Self {
            peekable,
            value: T::default(),
        }
    }
}