Exchange operator can take FnMut
@@ -22,12 +22,12 @@ pub trait Exchange<T, D: ExchangeData> {
/// .inspect(|x| println!("seen: {:?}", x));
/// });
/// ```
- fn exchange(&self, route: impl Fn(&D)->u64+'static) -> Self;
+ fn exchange(&self, route: impl FnMut(&D)->u64+'static) -> Self;
}
// impl<T: Timestamp, G: Scope<Timestamp=T>, D: ExchangeData> Exchange<T, D> for Stream<G, D> {
impl<G: Scope, D: ExchangeData> Exchange<G::Timestamp, D> for Stream<G, D> {
- fn exchange(&self, route: impl Fn(&D)->u64+'static) -> Stream<G, D> {
+ fn exchange(&self, route: impl FnMut(&D)->u64+'static) -> Stream<G, D> {
let mut vector = Default::default();
self.unary(ExchangePact::new(route), "Exchange", move |_,_| move |input, output| {
input.for_each(|time, data| {
