Method implement

src/plan/graphql_v2.rs:269–416

See Implementable::implement, as GraphQl v2 can't implement Implementable directly.

(
        &self,
        nested: &mut Iterative<'b, S, u64>,
        context: &mut I,
    )

    /// See Implementable::implement, as GraphQl v2 can't implement
    /// Implementable directly.
    pub fn implement<'b, T, I, S>(
        &self,
        nested: &mut Iterative<'b, S, u64>,
        context: &mut I,
    ) -> (Stream<S, Output>, ShutdownHandle)
    where
        T: Timestamp + Lattice,
        I: ImplContext<T>,
        S: Scope<Timestamp = T>,
        S::Timestamp: std::convert::Into<crate::timestamp::Time>,
    {
        use timely::dataflow::operators::Concatenate;

        let states = Rc::new(RefCell::new(JValue::Object(Map::new())));

        let dummy = VariableMap::new();

        let mut paths = {
            let mut paths_map = self
                .paths
                .iter()
                .flat_map(|path| {
                    let (streams, shutdown) = path.implement(nested, &dummy, context);
                    std::mem::forget(shutdown);
                    streams
                })
                .collect::<HashMap<PathId, _>>();

            let mut paths = paths_map.drain().collect::<Vec<(PathId, _)>>();

            // Important for cross-worker determinism.
            paths.sort_by_key(|(path_id, _)| path_id.clone());

            paths
        };

        let streams = paths.drain(..).map(|(path_id, stream)| {
            let states = states.clone();
            let mut buffer = HashMap::new();
            let mut vector = Vec::new();

            stream
                .exchange(|(path, _t, _diff)| path[0].clone().hashed())
                .delay(|(_path, t, _diff), _cap| t.clone())
                .unary_notify(
                    Pipeline,
                    "Changes",
                    vec![],
                    move |input, output, notificator| {
                        input.for_each(|cap, data| {
                            data.swap(&mut vector);
                            buffer
                                .entry(cap.time().clone())
                                .or_insert_with(Vec::new)
                                .extend(vector.drain(..));

                            notificator.notify_at(cap.retain());
                        });