generalize build
@@ -117,6 +117,23 @@ impl<G: Scope> OperatorBuilder<G> {
where
B: FnOnce(Vec<Capability<G::Timestamp>>) -> L,
L: FnMut(&[MutableAntichain<G::Timestamp>])+'static
+ {
+ self.build_reschedule(|caps| {
+ let mut logic = constructor(caps);
+ move |frontier| { logic(frontier); false }
+ })
+ }
+
+ /// Creates an operator implementation from supplied logic constructor.
+ ///
+ /// Unlike `build`, the supplied closure can indicate if the operator
+ /// should be considered incomplete. The `build` method indicates that
+ /// the operator is never incomplete and can be shut down at the system's
+ /// discretion.
+ pub fn build_reschedule<B, L>(self, constructor: B)
+ where
+ B: FnOnce(Vec<Capability<G::Timestamp>>) -> L,
+ L: FnMut(&[MutableAntichain<G::Timestamp>])->bool+'static
{
// create capabilities, discard references to their creation.
let mut capabilities = Vec::new();
@@ -143,7 +160,7 @@ impl<G: Scope> OperatorBuilder<G> {
}
// invoke supplied logic
- logic(&self_frontier[..]);
+ let result = logic(&self_frontier[..]);
// move batches of consumed changes.
for index in 0 .. progress.consumeds.len() {
@@ -162,7 +179,7 @@ impl<G: Scope> OperatorBuilder<G> {
self_produced[index].borrow_mut().drain_into(&mut progress.produceds[index]);
}
- false
+ result
};
self.builder.build(raw_logic);
