arrange example demonstrates shared queue
@@ -12,6 +12,7 @@ use rand::{Rng, SeedableRng, StdRng};
use differential_dataflow::AsCollection;
use differential_dataflow::operators::arrange::ArrangeByKey;
use differential_dataflow::trace::Cursor;
+use differential_dataflow::trace::Batch;
fn main() {
@@ -90,6 +91,8 @@ fn main() {
println!("Round {} finished after {:?}", next - 1, nanos);
let mut count = 0;
+
+ // we can directly interrogate the trace...
let timer = ::std::time::Instant::now();
let mut cursor = trace.cursor();
while cursor.key_valid() {
@@ -104,6 +107,17 @@ fn main() {
}
println!("count: {} in {:?}", count, timer.elapsed());
+
+ // or we can monitor data from a shared queue...
+ let queue = trace.new_listener();
+ let mut borrow = queue.borrow_mut();
+ while let Some((frontier, batch)) = borrow.pop_front() {
+ println!("received from queue:");
+ println!(" frontier: {:?}", frontier);
+ if let Some(batch) = batch {
+ println!(" batch from: {:?} -> {:?}, since: {:?}", batch.lower(), batch.upper(), batch.description().since());
+ }
+ }
}
}
}
