Trait Inspect 

pub trait Inspect<G: Scope, C>: InspectCore<G, C> + Sized
where
    for<'a> &'a C: IntoIterator,
{
    // Required method
    fn inspect_core<F>(self, func: F) -> Self
       where F: FnMut(Result<(&G::Timestamp, &C), &[G::Timestamp]>) + 'static;

    // Provided methods
    fn inspect<F>(self, func: F) -> Self
       where F: for<'a> FnMut(<&'a C as IntoIterator>::Item) + 'static { ... }
    fn inspect_time<F>(self, func: F) -> Self
       where F: for<'a> FnMut(&G::Timestamp, <&'a C as IntoIterator>::Item) + 'static { ... }
    fn inspect_batch(
        self,
        func: impl FnMut(&G::Timestamp, &C) + 'static,
    ) -> Self { ... }
}

Methods to inspect records and batches of records on a stream.

Required Methods

fn inspect_core<F>(self, func: F) -> Self

where F: FnMut(Result<(&G::Timestamp, &C), &[G::Timestamp]>) + 'static,

Runs a supplied closure on each observed data batch, and each frontier advancement.

Rust’s Result type is used to distinguish the events, with Ok for time and data, and Err for frontiers. Frontiers are only presented when they change.

Examples

use timely::dataflow::operators::{ToStream, Inspect};

timely::example(|scope| {
    (0..10).to_stream(scope)
           .container::<Vec<_>>()
           .inspect_core(|event| {
               match event {
                   Ok((time, data)) => println!("seen at: {:?}\t{:?} records", time, data.len()),
                   Err(frontier) => println!("frontier advanced to {:?}", frontier),
               }
            });
});

Provided Methods

fn inspect<F>(self, func: F) -> Self

where F: for<'a> FnMut(<&'a C as IntoIterator>::Item) + 'static,

Runs a supplied closure on each observed data element.

Examples

use timely::dataflow::operators::{ToStream, Inspect};

timely::example(|scope| {
    (0..10).to_stream(scope)
           .container::<Vec<_>>()
           .inspect(|x| println!("seen: {:?}", x));
});

Examples found in repository

examples/simple.rs (line 6)

fn main() {
    timely::example(|scope| {
        (0..10).to_stream(scope)
               .inspect(|x| println!("seen: {:?}", x));
    });
}

examples/threadless.rs (line 20)

fn main() {

    // create a naked single-threaded worker.
    let allocator = timely::communication::allocator::Thread::default();
    let mut worker = timely::worker::Worker::new(WorkerConfig::default(), allocator, None);

    // create input and probe handles.
    let mut input = InputHandle::new();
    let probe = ProbeHandle::new();

    // directly build a dataflow.
    worker.dataflow(|scope| {
        input
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x))
            .probe_with(&probe);
    });

    // manage inputs.
    for i in 0 .. 10 {
        input.send(i);
        input.advance_to(i);
        while probe.less_than(input.time()) {
            worker.step();
        }
    }
}

examples/capture_recv.rs (line 24)

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {

        let source_peers = std::env::args().nth(1).unwrap().parse::<usize>().unwrap();

        // create replayers from disjoint partition of source worker identifiers.
        let replayers =
        (0 .. source_peers)
            .filter(|i| i % worker.peers() == worker.index())
            .map(|i| TcpListener::bind(format!("127.0.0.1:{}", 8000 + i)).unwrap())
            .collect::<Vec<_>>()
            .into_iter()
            .map(|l| l.incoming().next().unwrap().unwrap())
            .map(EventReader::<_,Vec<u64>,_>::new)
            .collect::<Vec<_>>();

        worker.dataflow::<u64,_,_>(|scope| {
            replayers
                .replay_into(scope)
                .inspect(|x| println!("replayed: {:?}", x));
        })
    }).unwrap(); // asserts error-free execution
}

examples/rc.rs (line 21)

fn main() {
    // initializes and runs a timely dataflow.
    timely::execute_from_args(std::env::args(), |worker| {
        // create a new input, exchange data, and inspect its output
        let index = worker.index();
        let mut input = InputHandle::new();
        let probe = ProbeHandle::new();
        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                 .container::<Vec<_>>()
                 //.exchange(|x| *x) // <-- cannot exchange this; Rc is not Send.
                 .inspect(move |x| println!("worker {}:\thello {:?}", index, x))
                 .probe_with(&probe);
        });

        // introduce data and watch!
        for round in 0..10 {
            input.send(Test { _field: Rc::new(round) } );
            input.advance_to(round + 1);
            worker.step_while(|| probe.less_than(input.time()));
        }
    }).unwrap();
}

examples/hello.rs (line 17)

fn main() {
    // initializes and runs a timely dataflow.
    timely::execute_from_args(std::env::args(), |worker| {

        let index = worker.index();
        let mut input = InputHandle::new();
        let probe = ProbeHandle::new();

        // create a new input, exchange data, and inspect its output
        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                 .container::<Vec<_>>()
                 .exchange(|x| *x)
                 .inspect(move |x| println!("worker {}:\thello {}", index, x))
                 .probe_with(&probe);
        });

        // introduce data and watch!
        for round in 0..10 {
            if index == 0 {
                input.send(round);
            }
            input.advance_to(round + 1);
            while probe.less_than(input.time()) {
                worker.step();
            }
        }
    }).unwrap();
}

examples/logging-recv.rs (line 27)

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {

        let source_peers = std::env::args().nth(1).unwrap().parse::<usize>().unwrap();

        // create replayers from disjoint partition of source worker identifiers.
        let replayers =
        (0 .. source_peers)
            .filter(|i| i % worker.peers() == worker.index())
            .map(|i| TcpListener::bind(format!("127.0.0.1:{}", 8000 + i)).unwrap())
            .collect::<Vec<_>>()
            .into_iter()
            .map(|l| l.incoming().next().unwrap().unwrap())
            .map(EventReader::<Duration,Vec<(Duration,TimelySetup,TimelyEvent)>,_>::new)
            .collect::<Vec<_>>();

        worker.dataflow(|scope| {
            replayers
                .replay_into(scope)
                .inspect(|x| println!("replayed: {:?}", x));
        })
    }).unwrap(); // asserts error-free execution
}

Additional examples can be found in:

• examples/distinct.rs
• examples/wordcount.rs

fn inspect_time<F>(self, func: F) -> Self

where F: for<'a> FnMut(&G::Timestamp, <&'a C as IntoIterator>::Item) + 'static,

Runs a supplied closure on each observed data element and associated time.

Examples

use timely::dataflow::operators::{ToStream, Inspect};

timely::example(|scope| {
    (0..10).to_stream(scope)
           .container::<Vec<_>>()
           .inspect_time(|t, x| println!("seen at: {:?}\t{:?}", t, x));
});

Examples found in repository

examples/flow_controlled.rs (line 29)

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {
        let mut input = (0u64..100000).peekable();
        worker.dataflow(|scope| {
            let probe_handle = probe::Handle::new();
            let probe_handle_2 = probe_handle.clone();

            iterator_source(
                scope,
                "Source",
                move |prev_t| {
                    if let Some(first_x) = input.peek().cloned() {
                        let next_t = first_x / 100 * 100;
                        Some(IteratorSourceInput {
                            lower_bound: Default::default(),
                            data: vec![
                                (next_t,
                                 input.by_ref().take(10).map(|x| (/* "timestamp" */ x, x)).collect::<Vec<_>>())],
                            target: *prev_t,
                        })
                    } else {
                        None
                    }
                },
                probe_handle_2)
            .inspect_time(|t, d| eprintln!("@ {:?}: {:?}", t, d))
            .probe_with(&probe_handle);
        });
    }).unwrap();
}

fn inspect_batch(self, func: impl FnMut(&G::Timestamp, &C) + 'static) -> Self

Runs a supplied closure on each observed data batch (time and data slice).

Examples

use timely::dataflow::operators::{ToStream, Inspect};

timely::example(|scope| {
    (0..10).to_stream(scope)
           .container::<Vec<_>>()
           .inspect_batch(|t,xs| println!("seen at: {:?}\t{:?} records", t, xs.len()));
});

Examples found in repository

examples/unordered_input.rs (line 8)

fn main() {
    timely::execute(Config::thread(), |worker| {
        let (mut input, mut cap) = worker.dataflow::<usize,_,_>(|scope| {
            let (input, stream) = scope.new_unordered_input();
            stream.container::<Vec<_>>().inspect_batch(|t, x| println!("{:?} -> {:?}", t, x));
            input
        });

        for round in 0..10 {
            input.activate().session(&cap).give(round);
            cap = cap.delayed(&(round + 1));
            worker.step();
        }
    }).unwrap();
}

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors

impl<G: Scope, C: Container> Inspect<G, C> for Stream<G, C>

where for<'a> &'a C: IntoIterator,