timely/dataflow/operators/core/
inspect.rs

1//! Extension trait and implementation for observing and action on streamed data.
2
3use crate::Container;
4use crate::progress::{Stamp, Timestamp};
5use crate::dataflow::channels::pact::Pipeline;
6use crate::dataflow::Stream;
7use crate::dataflow::operators::generic::Operator;
8
9/// Methods to inspect records and container and frontier events on a stream.
10pub trait Inspect<T: Timestamp, C>: Sized {
11    /// Runs a supplied closure on each observed container, and each frontier advancement.
12    ///
13    /// Rust's `Result` type is used to distinguish the events, with `Ok` for a container and
14    /// the stamp under which it travels, and `Err` for frontiers. Frontiers are only presented
15    /// when they change. The stamp is the set of timestamps under which the container travels:
16    /// a singleton unless an upstream operator has stamped its messages with several times, or
17    /// with none.
18    ///
19    /// # Examples
20    /// ```
21    /// use timely::dataflow::operators::{ToStream, Inspect};
22    ///
23    /// timely::example(|scope| {
24    ///     (0..10).to_stream(scope)
25    ///            .container::<Vec<_>>()
26    ///            .inspect_core(|event| {
27    ///                match event {
28    ///                    Ok((stamp, data)) => println!("seen under {:?}: {:?} records", stamp, data.len()),
29    ///                    Err(frontier) => println!("frontier advanced to {:?}", frontier),
30    ///                }
31    ///             });
32    /// });
33    /// ```
34    fn inspect_core<F>(self, func: F) -> Self where F: FnMut(Result<(&Stamp<T>, &C), &[T]>)+'static;
35
36    /// Runs a supplied closure on each observed record.
37    ///
38    /// # Examples
39    /// ```
40    /// use timely::dataflow::operators::{ToStream, Inspect};
41    ///
42    /// timely::example(|scope| {
43    ///     (0..10).to_stream(scope)
44    ///            .container::<Vec<_>>()
45    ///            .inspect(|x| println!("seen: {:?}", x));
46    /// });
47    /// ```
48    fn inspect<F>(self, mut func: F) -> Self
49    where
50        for<'a> &'a C: IntoIterator,
51        F: for<'a> FnMut(<&'a C as IntoIterator>::Item) + 'static,
52    {
53        self.inspect_core(move |event| {
54            if let Ok((_, data)) = event {
55                for datum in data.into_iter() { func(datum); }
56            }
57        })
58    }
59}
60
61impl<T: Timestamp, C: Container> Inspect<T, C> for Stream<'_, T, C> {
62    fn inspect_core<F>(self, mut func: F) -> Self where F: FnMut(Result<(&Stamp<T>, &C), &[T]>)+'static {
63        let mut frontier = crate::progress::Antichain::from_elem(T::minimum());
64        self.unary_frontier(Pipeline, "Inspect", move |_,_| move |(input, chain), output| {
65            if chain.frontier() != frontier.borrow() {
66                frontier.clear();
67                frontier.extend(chain.frontier().iter().cloned());
68                func(Err(frontier.elements()));
69            }
70            input.for_each_stamp(|cap, data| {
71                let mut session = output.session(&cap);
72                for data in data {
73                    func(Ok((cap.stamp(), &*data)));
74                    session.give_container(data);
75                }
76            });
77        })
78    }
79}
timely/dataflow/operators/core/inspect.rs

timely/dataflow/operators/core/
inspect.rs
