Simpler container calls

Signed-off-by: Moritz Hoffmann <[email protected]>

Author: antiguru

timely/examples/wordcount.rs

@@ -17,13 +17,11 @@ fn main() {
         // create a new input, exchange data, and inspect its output
         worker.dataflow::<usize,_,_>(|scope| {
             input.to_stream(scope)
-                .container::<Vec<_>>()
                 .flat_map(|(text, diff): (String, i64)|
                     text.split_whitespace()
                         .map(move |word| (word.to_owned(), diff))
                         .collect::<Vec<_>>()
                  )
-                 .container::<Vec<_>>()
                  .unary_frontier(exchange, "WordCount", |_capability, _info| {
 
                     let mut queues = HashMap::new();

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

@@ -14,8 +14,8 @@ impl<T, D> PushOwned<T, D> {
     /// Create a new `PushOwned`. Similarly to `Tee`, it returns a pair where either element
     /// can be used as pusher or registrar.
     pub fn new() -> (Self, Self) {
-        let zelf = Self(Rc::new(RefCell::new(None)));
-        (zelf.clone(), zelf)
+        let shared = Rc::new(RefCell::new(None));
+        (Self(Rc::clone(&shared)), Self(shared))
     }
 
     /// Set the downstream pusher.
@@ -30,12 +30,6 @@ impl<T, D> fmt::Debug for PushOwned<T, D> {
     }
 }
 
-impl<T, D> Clone for PushOwned<T, D> {
-    fn clone(&self) -> Self {
-        Self(Rc::clone(&self.0))
-    }
-}
-
 impl<T, D> Push<Message<T, D>> for PushOwned<T, D> {
     #[inline]
     fn push(&mut self, message: &mut Option<Message<T, D>>) {

timely/src/dataflow/stream.rs

@@ -81,7 +81,7 @@ impl<S: Scope, C: Container + 'static> OwnedStream<S, C> {
     }
 
     /// Allows the assertion of a container type, for the benefit of type inference.
-    pub fn container<D: Container>(self) -> <Self as AsStream<S, D>>::Stream where Self: AsStream<S, D> { self.as_stream() }
+    pub fn container<D: Container>(self) -> OwnedStream<S, C> where Self: AsStream<S, D> { self.as_stream() }
 }
 
 /// A stream batching data in vectors.
@@ -136,24 +136,20 @@ impl<S: Scope, C: Container> StreamCore<S, C> {
     pub fn scope(&self) -> S { self.scope.clone() }
 
     /// Allows the assertion of a container type, for the benefit of type inference.
-    pub fn container<D: Container>(self) -> <Self as AsStream<S, D>>::Stream where Self: AsStream<S, D> { self.as_stream() }
+    pub fn container<D: Container>(self) -> StreamCore<S, C> where Self: AsStream<S, D> { self.as_stream() }
 }
 
 /// A type that can be translated to a [StreamCore].
 pub trait AsStream<S: Scope, C> {
-    /// The type of the stream.
-    type Stream;
     /// Translate `self` to a stream.
-    fn as_stream(self) -> Self::Stream;
+    fn as_stream(self) -> Self;
 }
 
 impl<S: Scope, C> AsStream<S, C> for StreamCore<S, C> {
-    type Stream = StreamCore<S, C>;
     fn as_stream(self) -> Self { self }
 }
 
 impl<G: Scope, C> AsStream<G, C> for OwnedStream<G, C> {
-    type Stream = OwnedStream<G, C>;
     fn as_stream(self) -> Self { self }
 }