Inspect tuples and associated times
@@ -19,6 +19,20 @@ pub trait Inspect<G: Scope, D: Data> {
/// });
/// ```
fn inspect<F: FnMut(&D)+'static>(&self, func: F) -> Self;
+
+ /// Runs a supplied closure on each observed data element and associated time.
+ ///
+ /// #Examples
+ /// ```
+ /// use timely::dataflow::operators::{ToStream, Map, Inspect};
+ ///
+ /// timely::example(|scope| {
+ /// (0..10).to_stream(scope)
+ /// .inspect_time(|t, x| println!("seen at: {:?}\t{:?}", t, x));
+ /// });
+ /// ```
+ fn inspect_time<F: FnMut(&G::Timestamp, &D)+'static>(&self, func: F) -> Self;
+
/// Runs a supplied closure on each observed data batch (time and data slice).
///
/// #Examples
@@ -44,6 +58,14 @@ impl<G: Scope, D: Data> Inspect<G, D> for Stream<G, D> {
})
}
+ fn inspect_time<F: FnMut(&G::Timestamp, &D)+'static>(&self, mut func: F) -> Stream<G, D> {
+ self.inspect_batch(move |time, data| {
+ for datum in data.iter() {
+ func(&time, &datum);
+ }
+ })
+ }
+
fn inspect_batch<F: FnMut(&G::Timestamp, &[D])+'static>(&self, mut func: F) -> Stream<G, D> {
self.unary_stream(Pipeline, "InspectBatch", move |input, output| {
input.for_each(|time, data| {
