impl Trait and default implementations in Inspect
@@ -6,7 +6,7 @@ use dataflow::{Stream, Scope};
use dataflow::operators::generic::unary::Unary;
/// Methods to inspect records and batches of records on a stream.
-pub trait Inspect<G: Scope, D: Data> {
+pub trait Inspect<G: Scope, D: Data>: Sized {
/// Runs a supplied closure on each observed data element.
///
/// #Examples
@@ -18,7 +18,11 @@ pub trait Inspect<G: Scope, D: Data> {
/// .inspect(|x| println!("seen: {:?}", x));
/// });
/// ```
- fn inspect<F: FnMut(&D)+'static>(&self, func: F) -> Self;
+ fn inspect(&self, mut func: impl FnMut(&D)+'static) -> Self {
+ self.inspect_batch(move |_, data| {
+ for datum in data.iter() { func(datum); }
+ })
+ }
/// Runs a supplied closure on each observed data element and associated time.
///
@@ -31,7 +35,13 @@ pub trait Inspect<G: Scope, D: Data> {
/// .inspect_time(|t, x| println!("seen at: {:?}\t{:?}", t, x));
/// });
/// ```
- fn inspect_time<F: FnMut(&G::Timestamp, &D)+'static>(&self, func: F) -> Self;
+ fn inspect_time(&self, mut func: impl FnMut(&G::Timestamp, &D)+'static) -> Self {
+ self.inspect_batch(move |time, data| {
+ for datum in data.iter() {
+ func(&time, &datum);
+ }
+ })
+ }
/// Runs a supplied closure on each observed data batch (time and data slice).
///
@@ -44,29 +54,11 @@ pub trait Inspect<G: Scope, D: Data> {
/// .inspect_batch(|t,xs| println!("seen at: {:?}\t{:?} records", t, xs.len()));
/// });
/// ```
- fn inspect_batch<F: FnMut(&G::Timestamp, &[D])+'static>(&self, func: F) -> Self;
+ fn inspect_batch(&self, func: impl FnMut(&G::Timestamp, &[D])+'static) -> Self;
}
impl<G: Scope, D: Data> Inspect<G, D> for Stream<G, D> {
-
- fn inspect<F: FnMut(&D)+'static>(&self, mut func: F) -> Stream<G, D> {
- self.unary_stream(Pipeline, "Inspect", move |input, output| {
- input.for_each(|time, data| {
- for datum in data.iter() { func(datum); }
- output.session(&time).give_content(data);
- });
- })
- }
-
- 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> {
+ fn inspect_batch(&self, mut func: impl FnMut(&G::Timestamp, &[D])+'static) -> Stream<G, D> {
self.unary_stream(Pipeline, "InspectBatch", move |input, output| {
input.for_each(|time, data| {
func(&time, &data[..]);
