timely/dataflow/operators/core/
to_stream.rs

1//! Conversion to the `Stream` type from iterators.
2
3use crate::container::{CapacityContainerBuilder, SizableContainer, PushInto};
4use crate::progress::Timestamp;
5use crate::{Container, ContainerBuilder};
6use crate::dataflow::operators::generic::operator::source;
7use crate::dataflow::{Stream, Scope};
8
9/// Converts to a timely [Stream], using a container builder.
10pub trait ToStreamBuilder<CB: ContainerBuilder> {
11    /// Converts to a timely [Stream], using the supplied container builder type.
12    ///
13    /// # Examples
14    ///
15    /// ```
16    /// use timely::dataflow::operators::core::{ToStreamBuilder, Capture};
17    /// use timely::dataflow::operators::core::capture::Extract;
18    /// use timely::container::CapacityContainerBuilder;
19    ///
20    /// let (data1, data2) = timely::example(|scope| {
21    ///     let data1 = ToStreamBuilder::<CapacityContainerBuilder<_>>::to_stream_with_builder(0..3, scope)
22    ///         .container::<Vec<_>>()
23    ///         .capture();
24    ///     let data2 = ToStreamBuilder::<CapacityContainerBuilder<_>>::to_stream_with_builder(vec![0,1,2], scope)
25    ///         .container::<Vec<_>>()
26    ///         .capture();
27    ///     (data1, data2)
28    /// });
29    ///
30    /// assert_eq!(data1.extract(), data2.extract());
31    /// ```
32    fn to_stream_with_builder<'scope, T: Timestamp>(self, scope: Scope<'scope, T>) -> Stream<'scope, T, CB::Container>;
33}
34
35impl<CB: ContainerBuilder, I: IntoIterator+'static> ToStreamBuilder<CB> for I where CB: PushInto<I::Item> {
36    fn to_stream_with_builder<'scope, T: Timestamp>(self, scope: Scope<'scope, T>) -> Stream<'scope, T, CB::Container> {
37
38        source::<_, CB, _, _>(scope, "ToStreamBuilder", |capability, info| {
39
40            // Acquire an activator, so that the operator can rescheduled itself.
41            let activator = scope.activator_for(info.address);
42
43            let mut iterator = self.into_iter().fuse();
44            let mut capability = Some(capability);
45
46            move |output| {
47
48                if let Some(element) = iterator.next() {
49                    let mut session = output.session_with_builder(capability.as_ref().unwrap());
50                    session.give(element);
51                    let n = 256 * crate::container::buffer::default_capacity::<I::Item>();
52                    session.give_iterator(iterator.by_ref().take(n - 1));
53                    activator.activate();
54                }
55                else {
56                    capability = None;
57                }
58            }
59        })
60    }
61}
62
63/// Converts to a timely [Stream]. Equivalent to [`ToStreamBuilder`] but
64/// uses a [`CapacityContainerBuilder`].
65pub trait ToStream<C> {
66    /// Converts to a timely [Stream].
67    ///
68    /// # Examples
69    ///
70    /// ```
71    /// use timely::dataflow::operators::core::{ToStream, Capture};
72    /// use timely::dataflow::operators::core::capture::Extract;
73    ///
74    /// let (data1, data2) = timely::example(|scope| {
75    ///     let data1 = (0..3).to_stream(scope).container::<Vec<_>>().capture();
76    ///     let data2 = vec![0,1,2].to_stream(scope).container::<Vec<_>>().capture();
77    ///     (data1, data2)
78    /// });
79    ///
80    /// assert_eq!(data1.extract(), data2.extract());
81    /// ```
82    fn to_stream<'scope, T: Timestamp>(self, scope: Scope<'scope, T>) -> Stream<'scope, T, C>;
83}
84
85impl<C: Container + SizableContainer, I: IntoIterator+'static> ToStream<C> for I where C: PushInto<I::Item> {
86    fn to_stream<'scope, T: Timestamp>(self, scope: Scope<'scope, T>) -> Stream<'scope, T, C> {
87        ToStreamBuilder::<CapacityContainerBuilder<C>>::to_stream_with_builder(self, scope)
88    }
89}
timely/dataflow/operators/core/to_stream.rs

timely/dataflow/operators/core/
to_stream.rs
