From 726179e182ea8820c43e0e2d0cc13725b8ce477f Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 26 Apr 2019 10:15:58 +0200
Subject: [PATCH] add Concatenate for streams

---
 timely/src/dataflow/operators/concat.rs | 19 +++++++++++++++++--
 1 file changed, 17 insertions(+), 2 deletions(-)

diff --git a/timely/src/dataflow/operators/concat.rs b/timely/src/dataflow/operators/concat.rs
index 9bd98e8a2..db5e0182a 100644
--- a/timely/src/dataflow/operators/concat.rs
+++ b/timely/src/dataflow/operators/concat.rs
@@ -47,11 +47,26 @@ pub trait Concatenate<G: Scope, D: Data> {
     ///          .inspect(|x| println!("seen: {:?}", x));
     /// });
     /// ```
-    fn concatenate(&self, _: impl IntoIterator<Item=Stream<G, D>>) -> Stream<G, D>;
+    fn concatenate<I>(&self, sources: I) -> Stream<G, D>
+    where
+        I: IntoIterator<Item=Stream<G, D>>;
+}
+
+impl<G: Scope, D: Data> Concatenate<G, D> for Stream<G, D> {
+    fn concatenate<I>(&self, sources: I) -> Stream<G, D>
+    where
+        I: IntoIterator<Item=Stream<G, D>>
+    {
+        let clone = self.clone();
+        self.scope().concatenate(Some(clone).into_iter().chain(sources))
+    }
 }
 
 impl<G: Scope, D: Data> Concatenate<G, D> for G {
-    fn concatenate(&self, sources: impl IntoIterator<Item=Stream<G, D>>) -> Stream<G, D> {
+    fn concatenate<I>(&self, sources: I) -> Stream<G, D>
+    where
+        I: IntoIterator<Item=Stream<G, D>>
+    {
 
         // create an operator builder.
         use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;