From 2b1f22ab714a6173fb439769e7391b9afc8327a2 Mon Sep 17 00:00:00 2001
From: David Selassie <david@bytewax.io>
Date: Fri, 21 Jan 2022 10:20:33 -0800
Subject: [PATCH] Exchange operator can take FnMut

---
 timely/src/dataflow/operators/exchange.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/timely/src/dataflow/operators/exchange.rs b/timely/src/dataflow/operators/exchange.rs
index 7ca26d402..878e4cd97 100644
--- a/timely/src/dataflow/operators/exchange.rs
+++ b/timely/src/dataflow/operators/exchange.rs
@@ -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| {