Test and relax bounds

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>

Author: antiguru

timely/src/dataflow/channels/pushers/buffer.rs

@@ -79,7 +79,7 @@ impl<T, C: Container, P: Push<BundleCore<T, C>>> BufferCore<T, C, P> where T: Eq
     }
 }
 
-impl<T, D: Data, P: Push<Bundle<T, D>>> Buffer<T, D, P> where T: Eq+Clone {
+impl<T, D: 'static, P: Push<Bundle<T, D>>> Buffer<T, D, P> where T: Eq+Clone {
     // internal method for use by `Session`.
     #[inline]
     fn give(&mut self, data: D) {
@@ -121,7 +121,7 @@ impl<'a, T, C: Container, P: Push<BundleCore<T, C>>+'a> Session<'a, T, C, P>  wh
     }
 }
 
-impl<'a, T, D: Data, P: Push<BundleCore<T, Vec<D>>>+'a> Session<'a, T, Vec<D>, P>  where T: Eq+Clone+'a, D: 'a {
+impl<'a, T, D: 'static, P: Push<BundleCore<T, Vec<D>>>+'a> Session<'a, T, Vec<D>, P>  where T: Eq+Clone+'a, D: 'a {
     /// Provides one record at the time specified by the `Session`.
     #[inline]
     pub fn give(&mut self, data: D) {

timely/src/dataflow/operators/to_stream.rs

@@ -12,7 +12,7 @@ use crate::progress::Timestamp;
 use crate::Data;
 
 /// Converts to a timely `Stream`.
-pub trait ToStream<T: Timestamp, D: Data> {
+pub trait ToStream<T: Timestamp, D> {
     /// Converts to a timely `Stream`.
     ///
     /// # Examples
@@ -32,7 +32,7 @@ pub trait ToStream<T: Timestamp, D: Data> {
     fn to_stream<S: Scope<Timestamp=T>>(self, scope: &mut S) -> OwnedStream<S, Vec<D>>;
 }
 
-impl<T: Timestamp, I: IntoIterator+'static> ToStream<T, I::Item> for I where I::Item: Data {
+impl<T: Timestamp, I: IntoIterator+'static> ToStream<T, I::Item> for I {
     fn to_stream<S: Scope<Timestamp=T>>(self, scope: &mut S) -> OwnedStream<S, Vec<I::Item>> {
 
         source(scope, "ToStream", |capability, info| {

timely/src/dataflow/stream.rs

@@ -126,3 +126,28 @@ where
             .finish()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::dataflow::channels::pact::Pipeline;
+    use crate::dataflow::operators::{Operator, ToStream};
+
+    #[derive(Debug, Eq, PartialEq)]
+    struct NotClone;
+
+    #[test]
+    fn test_non_clone_stream() {
+        crate::example(|scope| {
+            let _ = vec![NotClone]
+                .to_stream(scope)
+                .sink(Pipeline, "check non-clone", |input| {
+                    while let Some((_time, data)) = input.next() {
+                        for datum in data.iter() {
+                            assert_eq!(datum, &NotClone);
+                        }
+                    }
+                });
+        });
+    }
+
+}