pub fn observe_command(&mut self, command: &Command) {
            // Temporary storage for identifiers to add to and remove from frontier tracking.
            let mut start = Vec::new();
            let mut cease = Vec::new();
            command.frontier_tracking(&mut start, &mut cease);
            // Apply the determined effects of the command to `self.uppers`.
            for id in start.into_iter() {
                let mut frontier = timely::progress::frontier::MutableAntichain::new();
                frontier.update_iter(Some((
                    <Timestamp as timely::progress::Timestamp>::minimum(),
                    self.parts as i64,
                )));
                let previous = self.uppers.insert(id, frontier);
                assert!(previous.is_none(), "Protocol error: starting frontier tracking for already present identifier {:?} due to command {:?}", id, command);
            }
            for id in cease.into_iter() {
                let previous = self.uppers.remove(&id);
                if previous.is_none() {
                    log::debug!("Protocol error: ceasing frontier tracking for absent identifier {:?} due to command {:?}", id, command);
                }
            }
        }
/// Observes commands that move past, and prepares state for responses.
///
/// In particular, this method installs and removes upper frontier maintenance.
Observes commands that move past, and prepares state for responses.
In particular, this method installs and removes upper frontier maintenance.
Observes commands that move past and prepares state for responses . In particular this method installs and removes upper frontier maintenance .
pub fn observe_command(&mut self, command: &Command) {
            // Temporary storage for identifiers to add to and remove from frontier tracking.
            let mut start = Vec::new();
            let mut cease = Vec::new();
            command.frontier_tracking(&mut start, &mut cease);
            // Apply the determined effects of the command to `self.uppers`.
            for id in start.into_iter() {
                let mut frontier = timely::progress::frontier::MutableAntichain::new();
                frontier.update_iter(Some((
                    <Timestamp as timely::progress::Timestamp>::minimum(),
                    self.parts as i64,
                )));
                let previous = self.uppers.insert(id, frontier);
                assert!(previous.is_none(), "Protocol error: starting frontier tracking for already present identifier {:?} due to command {:?}", id, command);
            }
            for id in cease.into_iter() {
                let previous = self.uppers.remove(&id);
                if previous.is_none() {
                    log::debug!("Protocol error: ceasing frontier tracking for absent identifier {:?} due to command {:?}", id, command);
                }
            }
        }
pub fn observe_command ( & mut self , command : & Command ) { // Temporary storage for identifiers to add to and remove from frontier tracking. let mut start = Vec :: new ( ) ; let mut cease = Vec :: new ( ) ; command . frontier_tracking ( & mut start , & mut cease ) ; // Apply the determined effects of the command to `self.uppers`. for id in start . into_iter ( ) { let mut frontier = timely :: progress :: frontier :: MutableAntichain :: new ( ) ; frontier . update_iter ( Some ( ( < Timestamp as timely :: progress :: Timestamp > :: minimum ( ) , self . parts as i64 , ) ) ) ; let previous = self . uppers . insert ( id , frontier ) ; assert ! ( previous is_none ( ) " " id command ) ; } for id in cease . into_iter ( ) { let previous = self . uppers . remove ( & id ) ; if previous . is_none ( ) { log :: debug ! ( " " id command ) ; } } }
Observes commands that move past, and prepares state for responses.
Observes commands that move past and prepares state for responses .
&mut self
command
&Command
&mut self
command
&Command