Function iterator_operator

src/storage-operators/src/persist_source.rs:1434–1463

An operator that emits `Part`'s at the specified timestamps. Does not drop its capability until it gets a signal from the `Sender` it returns.

(
        scope: Scope<'scope, (u64, Subtime)>,
        mut input: I,
    )

    /// An operator that emits `Part`'s at the specified timestamps. Does not
    /// drop its capability until it gets a signal from the `Sender` it returns.
    fn iterator_operator<'scope, I: Iterator<Item = (u64, Part)> + 'static>(
        scope: Scope<'scope, (u64, Subtime)>,
        mut input: I,
    ) -> (StreamVec<'scope, (u64, Subtime), Part>, oneshot::Sender<()>) {
        let (finalizer_tx, finalizer_rx) = oneshot::channel();
        let mut iterator = AsyncOperatorBuilder::new("iterator".to_string(), scope);
        let (output_handle, output) = iterator.new_output::<CapacityContainerBuilder<Vec<Part>>>();

        iterator.build(|mut caps| async move {
            let mut capability = Some(caps.pop().unwrap());
            let mut last = None;
            while let Some(element) = input.next() {
                let time = element.0.clone();
                let part = element.1;
                last = Some((time, Subtime(0)));
                output_handle.give(&capability.as_ref().unwrap().delayed(&last.unwrap()), part);
            }
            if let Some(last) = last {
                capability
                    .as_mut()
                    .unwrap()
                    .downgrade(&(last.0 + 1, last.1));
            }

            let _ = finalizer_rx.await;
            capability.take();
        });

        (output, finalizer_tx)
    }

    /// An operator that consumes its input ONLY when given a signal to do from
    /// the `UnboundedSender` it returns. Each `send` corresponds with 1 `Data` event