operator source
@@ -261,7 +261,73 @@ fn binary_base<G: Scope, D1: Data, D2: Data, D3: Data, P1, P2>(
(targets, registrar), scope)
}
+/// Creates a new data stream source for a scope.
+///
+/// The source is defined by a name, and a constructor which takes a default capability to
+/// a method that can be repeatedly called on a output handle. The method is then repeatedly
+/// invoked, and is expected to eventually send data and downgrade and release capabilities.
+///
+/// #Examples
+/// ```
+/// use timely::dataflow::operators::Inspect;
+/// use timely::dataflow::operators::operator::source;
+/// use timely::dataflow::Scope;
+///
+/// timely::example(|scope| {
+///
+/// source(scope, "Source", |capability| {
+/// let mut cap = Some(capability);
+/// move |output| {
+///
+/// let mut done = false;
+/// if let Some(cap) = cap.as_mut() {
+/// // get some data and send it.
+/// let mut time = cap.time();
+/// output.session(&cap)
+/// .give(cap.time().inner);
+///
+/// // downgrade capability.
+/// time.inner += 1;
+/// *cap = cap.delayed(&time);
+/// done = time.inner > 20;
+/// }
+///
+/// if done { cap = None; }
+/// }
+/// })
+/// .inspect(|x| println!("number: {:?}", x));
+/// });
+/// ```
+pub fn source<G: Scope, D, B, L>(scope: &G, name: &str, constructor: B) -> Stream<G, D>
+where
+ D: Data,
+ B: Fn(Capability<G::Timestamp>) -> L,
+ L: FnMut(&mut OutputHandle<G::Timestamp, D, Tee<G::Timestamp, D>>)+'static {
+
+ let (cap, internal_changes) = make_default_cap::<G>();
+ let mut logic = constructor(cap);
+
+ let (targets, registrar) = Tee::<G::Timestamp,D>::new();
+
+ let operator = OperatorImpl::new(
+ name.to_owned(),
+ 0,
+ scope.peers(),
+ move |_consumed, _internal, _frontiers, output| {
+ logic(output);
+ },
+ internal_changes,
+ targets,
+ false
+ );
+
+ let mut scope: G = scope.clone();
+ let index = scope.add_operator(operator);
+ Stream::new(Source { index: index, port: 0 }, registrar, scope)
+}
+
impl<G: Scope, D1: Data> Operator<G, D1> for Stream<G, D1> {
+
fn unary_frontier<D2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> Stream<G, D2>
where
D2: Data,
