fn shutdown_diff(&mut self, diff: &ConfigDiff, rt: &mut runtime::Runtime) {
        // Sources
        let timeout = Duration::from_secs(30); //sec

        // First pass to tell the sources to shut down.
        let mut source_shutdown_complete_futures = Vec::new();

        // Only log that we are waiting for shutdown if we are actually removing
        // sources.
        if !diff.sources.to_remove.is_empty() {
            info!(
                "Waiting for up to {} seconds for sources to finish shutting down",
                timeout.as_secs()
            );
        }

        let deadline = Instant::now() + timeout;
        for name in &diff.sources.to_remove {
            info!("Removing source {:?}", name);

            self.tasks.remove(name).unwrap().forget();

            self.remove_outputs(name);
            source_shutdown_complete_futures
                .push(self.shutdown_coordinator.shutdown_source(name, deadline));
        }
        for name in &diff.sources.to_change {
            self.remove_outputs(name);
            source_shutdown_complete_futures
                .push(self.shutdown_coordinator.shutdown_source(name, deadline));
        }

        // Wait for the shutdowns to complete

        // Only log message if there are actual futures to check.
        if !source_shutdown_complete_futures.is_empty() {
            info!(
                "Waiting for up to {} seconds for sources to finish shutting down",
                timeout.as_secs()
            );
        }

        rt.block_on(future::join_all(source_shutdown_complete_futures))
            .unwrap();

        // Second pass now that all sources have shut down for final cleanup.
        for name in &diff.sources.to_remove {
            self.source_tasks.remove(name).wait().unwrap();
        }
        for name in &diff.sources.to_change {
            self.source_tasks.remove(name).wait().unwrap();
        }

        // Transforms
        for name in &diff.transforms.to_remove {
            info!("Removing transform {:?}", name);

            self.tasks.remove(name).unwrap().forget();

            self.remove_inputs(&name);
            self.remove_outputs(&name);
        }

        // Sinks
        for name in &diff.sinks.to_remove {
            info!("Removing sink {:?}", name);

            self.tasks.remove(name).unwrap().forget();

            self.remove_inputs(&name);
        }
    }
/// Shutdowns removed and replaced pieces of topology.
Shutdowns removed and replaced pieces of topology.
Shutdowns removed and replaced pieces of topology .
fn shutdown_diff(&mut self, diff: &ConfigDiff, rt: &mut runtime::Runtime) {
        // Sources
        let timeout = Duration::from_secs(30); //sec
        // First pass to tell the sources to shut down.
        let mut source_shutdown_complete_futures = Vec::new();
        // Only log that we are waiting for shutdown if we are actually removing
        // sources.
        if !diff.sources.to_remove.is_empty() {
            info!(
                "Waiting for up to {} seconds for sources to finish shutting down",
                timeout.as_secs()
            );
        }
        let deadline = Instant::now() + timeout;
        for name in &diff.sources.to_remove {
            info!("Removing source {:?}", name);
            self.tasks.remove(name).unwrap().forget();
            self.remove_outputs(name);
            source_shutdown_complete_futures
                .push(self.shutdown_coordinator.shutdown_source(name, deadline));
        }
        for name in &diff.sources.to_change {
            self.remove_outputs(name);
            source_shutdown_complete_futures
                .push(self.shutdown_coordinator.shutdown_source(name, deadline));
        }
        // Wait for the shutdowns to complete
        // Only log message if there are actual futures to check.
        if !source_shutdown_complete_futures.is_empty() {
            info!(
                "Waiting for up to {} seconds for sources to finish shutting down",
                timeout.as_secs()
            );
        }
        rt.block_on(future::join_all(source_shutdown_complete_futures))
            .unwrap();
        // Second pass now that all sources have shut down for final cleanup.
        for name in &diff.sources.to_remove {
            self.source_tasks.remove(name).wait().unwrap();
        }
        for name in &diff.sources.to_change {
            self.source_tasks.remove(name).wait().unwrap();
        }
        // Transforms
        for name in &diff.transforms.to_remove {
            info!("Removing transform {:?}", name);
            self.tasks.remove(name).unwrap().forget();
            self.remove_inputs(&name);
            self.remove_outputs(&name);
        }
        // Sinks
        for name in &diff.sinks.to_remove {
            info!("Removing sink {:?}", name);
            self.tasks.remove(name).unwrap().forget();
            self.remove_inputs(&name);
        }
    }
fn shutdown_diff ( & mut self , diff : & ConfigDiff , rt : & mut runtime :: Runtime ) { // Sources let timeout = Duration :: from_secs ( 30 ) ; //sec // First pass to tell the sources to shut down. let mut source_shutdown_complete_futures = Vec :: new ( ) ; // Only log that we are waiting for shutdown if we are actually removing // sources. if ! diff . sources . to_remove . is_empty ( ) { info ! ( " " timeout as_secs ( ) ) ; } let deadline = Instant :: now ( ) + timeout ; for name in & diff . sources . to_remove { info ! ( " " name ) ; self . tasks . remove ( name ) . unwrap ( ) . forget ( ) ; self . remove_outputs ( name ) ; source_shutdown_complete_futures . push ( self . shutdown_coordinator . shutdown_source ( name , deadline ) ) ; } for name in & diff . sources . to_change { self . remove_outputs ( name ) ; source_shutdown_complete_futures . push ( self . shutdown_coordinator . shutdown_source ( name , deadline ) ) ; } // Wait for the shutdowns to complete // Only log message if there are actual futures to check. if ! source_shutdown_complete_futures . is_empty ( ) { info ! ( " " timeout as_secs ( ) ) ; } rt . block_on ( future :: join_all ( source_shutdown_complete_futures ) ) . unwrap ( ) ; // Second pass now that all sources have shut down for final cleanup. for name in & diff . sources . to_remove { self . source_tasks . remove ( name ) . wait ( ) . unwrap ( ) ; } for name in & diff . sources . to_change { self . source_tasks . remove ( name ) . wait ( ) . unwrap ( ) ; } // Transforms for name in & diff . transforms . to_remove { info ! ( " " name ) ; self . tasks . remove ( name ) . unwrap ( ) . forget ( ) ; self . remove_inputs ( & name ) ; self . remove_outputs ( & name ) ; } // Sinks for name in & diff . sinks . to_remove { info ! ( " " name ) ; self . tasks . remove ( name ) . unwrap ( ) . forget ( ) ; self . remove_inputs ( & name ) ; } }
Shutdowns removed and replaced pieces of topology.
Shutdowns removed and replaced pieces of topology .
&mut self
diff
&ConfigDiff
rt
&mut runtime::Runtime