Method register_source

src/server/mod.rs:433–479

Handle a RegisterSource request.

(
        &mut self,
        source: Box<dyn Sourceable<S>>,
        scope: &mut S,
    )

    /// Handle a RegisterSource request.
    pub fn register_source<S: Scope<Timestamp = T>>(
        &mut self,
        source: Box<dyn Sourceable<S>>,
        scope: &mut S,
    ) -> Result<(), Error> {
        // use timely::logging::Logger;
        // let timely_logger = scope.log_register().remove("timely");

        // let differential_logger = scope.log_register().remove("differential/arrange");

        let context = SourcingContext {
            t0: self.t0,
            scheduler: Rc::downgrade(&self.scheduler),
            domain_probe: self.context.internal.domain_probe().clone(),
            timely_events: self.timely_events.clone().unwrap(),
            differential_events: self.differential_events.clone().unwrap(),
        };

        // self.timely_events = None;
        // self.differential_events = None;

        let mut attribute_streams = source.source(scope, context);

        for (aid, config, datoms) in attribute_streams.drain(..) {
            self.context
                .internal
                .create_sourced_attribute(&aid, config, &datoms)?;
        }

        // if let Some(logger) = timely_logger {
        //     if let Ok(logger) = logger.downcast::<Logger<TimelyEvent>>() {
        //         scope
        //             .log_register()
        //             .insert_logger::<TimelyEvent>("timely", *logger);
        //     }
        // }

        // if let Some(logger) = differential_logger {
        //     if let Ok(logger) = logger.downcast::<Logger<DifferentialEvent>>() {
        //         scope
        //             .log_register()
        //             .insert_logger::<DifferentialEvent>("differential/arrange", *logger);
        //     }
        // }

        Ok(())
    }

    /// Handle an AdvanceDomain request.
    pub fn advance_domain(&mut self, name: Option<String>, next: T) -> Result<(), Error> {