timely/dataflow/
stream.rs

1//! A handle to a typed stream of timely data.
2//!
3//! Most high-level timely dataflow programming is done with streams, which are each a handle to an
4//! operator output. Extension methods on the `Stream` type provide the appearance of higher-level
5//! declarative programming, while constructing a dataflow graph underneath.
6
7use crate::progress::{Source, Target};
8
9use crate::communication::Push;
10use crate::dataflow::Scope;
11use crate::dataflow::channels::pushers::tee::TeeHelper;
12use crate::dataflow::channels::Message;
13use std::fmt::{self, Debug};
14
15// use dataflow::scopes::root::loggers::CHANNELS_Q;
16
17/// Abstraction of a stream of `C: Container` records timestamped with `S::Timestamp`.
18///
19/// Internally `Stream` maintains a list of data recipients who should be presented with data
20/// produced by the source of the stream.
21pub struct Stream<S: Scope, C> {
22    /// The progress identifier of the stream's data source.
23    name: Source,
24    /// The `Scope` containing the stream.
25    scope: S,
26    /// Maintains a list of Push<Message<T, C>> interested in the stream's output.
27    ports: TeeHelper<S::Timestamp, C>,
28}
29
30impl<S: Scope, C: Clone+'static> Clone for Stream<S, C> {
31    fn clone(&self) -> Self {
32        Self {
33            name: self.name,
34            scope: self.scope.clone(),
35            ports: self.ports.clone(),
36        }
37    }
38
39    fn clone_from(&mut self, source: &Self) {
40        self.name.clone_from(&source.name);
41        self.scope.clone_from(&source.scope);
42        self.ports.clone_from(&source.ports);
43    }
44}
45
46/// A stream batching data in owning vectors.
47pub type StreamVec<S, D> = Stream<S, Vec<D>>;
48
49impl<S: Scope, C> Stream<S, C> {
50    /// Connects the stream to a destination.
51    ///
52    /// The destination is described both by a `Target`, for progress tracking information, and a `P: Push` where the
53    /// records should actually be sent. The identifier is unique to the edge and is used only for logging purposes.
54    pub fn connect_to<P: Push<Message<S::Timestamp, C>>+'static>(self, target: Target, pusher: P, identifier: usize) where C: 'static {
55
56        let mut logging = self.scope().logging();
57        logging.as_mut().map(|l| l.log(crate::logging::ChannelsEvent {
58            id: identifier,
59            scope_addr: self.scope.addr().to_vec(),
60            source: (self.name.node, self.name.port),
61            target: (target.node, target.port),
62            typ: std::any::type_name::<C>().to_string(),
63        }));
64
65        self.scope.add_edge(self.name, target);
66        self.ports.add_pusher(pusher);
67    }
68    /// Allocates a `Stream` from a supplied `Source` name and rendezvous point.
69    pub fn new(source: Source, output: TeeHelper<S::Timestamp, C>, scope: S) -> Self {
70        Self { name: source, ports: output, scope }
71    }
72    /// The name of the stream's source operator.
73    pub fn name(&self) -> &Source { &self.name }
74    /// The scope immediately containing the stream.
75    pub fn scope(&self) -> S { self.scope.clone() }
76
77    /// Allows the assertion of a container type, for the benefit of type inference.
78    ///
79    /// This method can be needed when the container type of a stream is unconstrained,
80    /// most commonly after creating an input, or bracking wholly generic operators.
81    ///
82    /// # Examples
83    /// ```
84    /// use timely::dataflow::operators::{ToStream, Inspect};
85    ///
86    /// timely::example(|scope| {
87    ///     (0..10).to_stream(scope)
88    ///            .container::<Vec<_>>()
89    ///            .inspect(|x| println!("seen: {:?}", x));
90    /// });
91    /// ```
92    pub fn container<C2>(self) -> Stream<S, C2> where Self: AsStream<S, C2> { self.as_stream() }
93}
94
95/// A type that can be translated to a [Stream].
96pub trait AsStream<S: Scope, C> {
97    /// Translate `self` to a [Stream].
98    fn as_stream(self) -> Stream<S, C>;
99}
100
101impl<S: Scope, C> AsStream<S, C> for Stream<S, C> {
102    fn as_stream(self) -> Self { self }
103}
104
105impl<S, C> Debug for Stream<S, C>
106where
107    S: Scope,
108{
109    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
110        f.debug_struct("Stream")
111            .field("source", &self.name)
112            .finish_non_exhaustive()
113    }
114}
115
116#[cfg(test)]
117mod tests {
118    use crate::dataflow::channels::pact::Pipeline;
119    use crate::dataflow::operators::{Operator, ToStream};
120
121    #[derive(Debug, Eq, PartialEq)]
122    struct NotClone;
123
124    #[test]
125    fn test_non_clone_stream() {
126        crate::example(|scope| {
127            let _ = [NotClone]
128                .to_stream(scope)
129                .container::<Vec<_>>()
130                .sink(Pipeline, "check non-clone", |(input, _frontier)| {
131                    input.for_each(|_cap, data| {
132                        for datum in data.drain(..) {
133                            assert_eq!(datum, NotClone);
134                        }
135                    });
136                });
137        });
138    }
139}
timely/dataflow/stream.rs

timely/dataflow/
stream.rs
