async fn drain_skip_uppers(&mut self, exclude_uppers: &HashSet<GlobalId>) {
        self.drain_feedback_msgs().await;
        let mut to_send = vec![];
        let mut to_queue = vec![];
        for mut msg in self.queued_feedback.drain(..) {
            // Filter out requested ids.
            if let dataflow_types::client::Response::FrontierUppers(uppers) = &mut msg {
                // Requeue excluded uppers so future wait-sql directives don't always have to
                // specify the same exclude list forever.
                let mut requeue = uppers.clone();
                requeue.retain(|(id, _data)| exclude_uppers.contains(id));
                if !requeue.is_empty() {
                    to_queue.push(dataflow_types::client::Response::FrontierUppers(requeue));
                }
                uppers.retain(|(id, _data)| !exclude_uppers.contains(id));
            }
            to_send.push(msg);
        }
        for msg in to_send {
            self.dataflow_client.forward_response(msg);
        }
        self.queued_feedback = to_queue;
    }
// Drains messages from the queue into coord, extracting and requeueing
// excluded uppers.
Drains messages from the queue into coord, extracting and requeueing
excluded uppers.
Drains messages from the queue into coord extracting and requeueing excluded uppers .
async fn drain_skip_uppers(&mut self, exclude_uppers: &HashSet<GlobalId>) {
        self.drain_feedback_msgs().await;
        let mut to_send = vec![];
        let mut to_queue = vec![];
        for mut msg in self.queued_feedback.drain(..) {
            // Filter out requested ids.
            if let dataflow_types::client::Response::FrontierUppers(uppers) = &mut msg {
                // Requeue excluded uppers so future wait-sql directives don't always have to
                // specify the same exclude list forever.
                let mut requeue = uppers.clone();
                requeue.retain(|(id, _data)| exclude_uppers.contains(id));
                if !requeue.is_empty() {
                    to_queue.push(dataflow_types::client::Response::FrontierUppers(requeue));
                }
                uppers.retain(|(id, _data)| !exclude_uppers.contains(id));
            }
            to_send.push(msg);
        }
        for msg in to_send {
            self.dataflow_client.forward_response(msg);
        }
        self.queued_feedback = to_queue;
    }
async fn drain_skip_uppers ( & mut self , exclude_uppers : & HashSet < GlobalId > ) { self . drain_feedback_msgs ( ) . await ; let mut to_send = vec ! [ ] ; let mut to_queue = vec ! [ ] ; for mut msg in self . queued_feedback . drain ( .. ) { // Filter out requested ids. if let dataflow_types :: client :: Response :: FrontierUppers ( uppers ) = & mut msg { // Requeue excluded uppers so future wait-sql directives don't always have to // specify the same exclude list forever. let mut requeue = uppers . clone ( ) ; requeue . retain ( | ( id , _data ) | exclude_uppers . contains ( id ) ) ; if ! requeue . is_empty ( ) { to_queue . push ( dataflow_types :: client :: Response :: FrontierUppers ( requeue ) ) ; } uppers . retain ( | ( id , _data ) | ! exclude_uppers . contains ( id ) ) ; } to_send . push ( msg ) ; } for msg in to_send { self . dataflow_client . forward_response ( msg ) ; } self . queued_feedback = to_queue ; }
Drains messages from the queue into coord, extracting and requeueing
excluded uppers.
Drains messages from the queue into coord extracting and requeueing excluded uppers .
&mut self
exclude_uppers
&HashSet<GlobalId>
&mut self
exclude_uppers
&HashSet<GlobalId>