Remove unneeded Sink trait

src/operator/sink/collect.rs

@@ -2,7 +2,7 @@ use std::fmt::Display;
 use std::marker::PhantomData;
 
 use crate::block::{BlockStructure, OperatorKind, OperatorStructure};
-use crate::operator::sink::{Sink, StreamOutputRef};
+use crate::operator::sink::StreamOutputRef;
 use crate::operator::{ExchangeData, Operator, StreamElement};
 use crate::scheduler::ExecutionMetadata;
 
@@ -77,13 +77,6 @@ where
     }
 }
 
-impl<Out: ExchangeData, C: FromIterator<Out> + Send, PreviousOperators> Sink
-    for Collect<Out, C, PreviousOperators>
-where
-    PreviousOperators: Operator<Out = Out>,
-{
-}
-
 impl<Out: ExchangeData, C: FromIterator<Out> + Send, PreviousOperators> Clone
     for Collect<Out, C, PreviousOperators>
 where

src/operator/sink/collect_channel.rs

@@ -1,7 +1,6 @@
 use std::fmt::Display;
 
 use crate::block::{BlockStructure, OperatorKind, OperatorStructure};
-use crate::operator::sink::Sink;
 use crate::operator::{ExchangeData, Operator, StreamElement};
 use crate::scheduler::ExecutionMetadata;
 
@@ -67,11 +66,6 @@ where
     }
 }
 
-impl<Out: ExchangeData, PreviousOperators> Sink for CollectChannelSink<Out, PreviousOperators> where
-    PreviousOperators: Operator<Out = Out>
-{
-}
-
 #[cfg(test)]
 mod tests {
     use itertools::Itertools;

src/operator/sink/collect_count.rs

@@ -2,7 +2,7 @@ use std::fmt::Display;
 
 use crate::block::{BlockStructure, OperatorKind, OperatorStructure};
 
-use crate::operator::sink::{Sink, StreamOutputRef};
+use crate::operator::sink::StreamOutputRef;
 use crate::operator::{Operator, StreamElement};
 use crate::scheduler::ExecutionMetadata;
 
@@ -71,11 +71,6 @@ where
     }
 }
 
-impl<PreviousOperators> Sink for CollectCountSink<PreviousOperators> where
-    PreviousOperators: Operator<Out = usize>
-{
-}
-
 impl<PreviousOperators> Clone for CollectCountSink<PreviousOperators>
 where
     PreviousOperators: Operator<Out = usize>,

src/operator/sink/collect_vec.rs

@@ -1,7 +1,7 @@
 use std::fmt::Display;
 
 use crate::block::{BlockStructure, OperatorKind, OperatorStructure};
-use crate::operator::sink::{Sink, StreamOutputRef};
+use crate::operator::sink::StreamOutputRef;
 use crate::operator::{ExchangeData, Operator, StreamElement};
 use crate::scheduler::ExecutionMetadata;
 
@@ -75,11 +75,6 @@ where
     }
 }
 
-impl<Out: ExchangeData, PreviousOperators> Sink for CollectVecSink<Out, PreviousOperators> where
-    PreviousOperators: Operator<Out = Out>
-{
-}
-
 impl<Out: ExchangeData, PreviousOperators> Clone for CollectVecSink<Out, PreviousOperators>
 where
     PreviousOperators: Operator<Out = Out>,

src/operator/sink/mod.rs

@@ -5,30 +5,23 @@
 
 use std::sync::{Arc, Mutex};
 
-use crate::operator::Operator;
-
 pub(super) mod collect;
 pub(super) mod collect_channel;
 pub(super) mod collect_count;
 pub(super) mod collect_vec;
 pub(super) mod for_each;
 
-/// This trait marks all the operators that can be used as sinks.
-pub(crate) trait Sink: Operator<Out = ()> {}
-
 pub(crate) type StreamOutputRef<Out> = Arc<Mutex<Option<Out>>>;
 
 /// The result of a stream after the execution.
 ///
 /// This will eventually hold the value _after_ the environment has been fully executed. To access
 /// the content of the output you have to call [`StreamOutput::get`].
-pub struct StreamOutput<Out> {
-    result: StreamOutputRef<Out>,
-}
+pub struct StreamOutput<Out>(StreamOutputRef<Out>);
 
 impl<Out> From<StreamOutputRef<Out>> for StreamOutput<Out> {
     fn from(value: StreamOutputRef<Out>) -> Self {
-        Self { result: value }
+        Self(value)
     }
 }
 
@@ -38,9 +31,6 @@ impl<Out> StreamOutput<Out> {
     /// This will consume the result and return the owned content. If the content has already been
     /// extracted, or if the content is not ready yet, this will return `None`.
     pub fn get(self) -> Option<Out> {
-        self.result
-            .try_lock()
-            .expect("Cannot lock output result")
-            .take()
+        self.0.try_lock().expect("Cannot lock output result").take()
     }
 }

src/operator/start/binary.rs

@@ -324,16 +324,16 @@ impl<OutL: ExchangeData, OutR: ExchangeData> StartReceiver for BinaryStartReceiv
         previous
     }
 
-    fn cached_replicas(&self) -> usize {
-        let mut cached = 0;
-        if self.left.cached {
-            cached += self.left.instances
-        }
-        if self.right.cached {
-            cached += self.right.instances
-        }
-        cached
-    }
+    // fn cached_replicas(&self) -> usize {
+    //     let mut cached = 0;
+    //     if self.left.cached {
+    //         cached += self.left.instances
+    //     }
+    //     if self.right.cached {
+    //         cached += self.right.instances
+    //     }
+    //     cached
+    // }
 
     fn recv_timeout(
         &mut self,

src/operator/start/mod.rs

@@ -31,9 +31,9 @@ pub(crate) trait StartReceiver: Clone {
     /// This list should contain all the replicas this receiver will receive data from.
     fn prev_replicas(&self) -> Vec<Coord>;
 
-    /// The number of those replicas which are behind a cache, and therefore never will emit a
-    /// `StreamElement::Terminate` message.
-    fn cached_replicas(&self) -> usize;
+    // /// The number of those replicas which are behind a cache, and therefore never will emit a
+    // /// `StreamElement::Terminate` message.
+    // fn cached_replicas(&self) -> usize;
 
     /// Try to receive a batch from the previous blocks, or fail with an error if the timeout
     /// expires.

src/operator/start/simple.rs

@@ -51,10 +51,6 @@ impl<Out: ExchangeData> StartReceiver for SimpleStartReceiver<Out> {
         self.previous_replicas.clone()
     }
 
-    fn cached_replicas(&self) -> usize {
-        0
-    }
-
     fn recv_timeout(&mut self, timeout: Duration) -> Result<NetworkMessage<Out>, RecvTimeoutError> {
         let receiver = self.receiver.as_mut().unwrap();
         receiver.recv_timeout(timeout)

src/operator/window/aggr/collect_vec.rs

@@ -52,4 +52,13 @@ where
         };
         self.add_window_operator("WindowMap", acc)
     }
+
+    pub fn to_vec(self) -> KeyedStream<impl Operator<Out = (Key, Vec<Out>)>> {
+        let acc = CollectVec::<Out, Vec<Out>, _> {
+            vec: Default::default(),
+            f: |v| v,
+            _o: PhantomData,
+        };
+        self.add_window_operator("WindowMap", acc)
+    }
 }