async fn drain_peek_response(&mut self) {
        self.drain_feedback_msgs().await;
        let mut to_send = vec![];
        let mut to_queue = vec![];
        for msg in self.queued_feedback.drain(..) {
            if let dataflow_types::client::Response::PeekResponse(..) = msg {
                to_send.push(msg);
            } else {
                to_queue.push(msg);
            }
        }
        for msg in to_send {
            self.dataflow_client.forward_response(msg);
        }
        self.queued_feedback = to_queue;
    }
// Drains PeekResponse messages from the queue into coord.
Drains PeekResponse messages from the queue into coord.
Drains PeekResponse messages from the queue into coord .
async fn drain_peek_response(&mut self) {
        self.drain_feedback_msgs().await;
        let mut to_send = vec![];
        let mut to_queue = vec![];
        for msg in self.queued_feedback.drain(..) {
            if let dataflow_types::client::Response::PeekResponse(..) = msg {
                to_send.push(msg);
            } else {
                to_queue.push(msg);
            }
        }
        for msg in to_send {
            self.dataflow_client.forward_response(msg);
        }
        self.queued_feedback = to_queue;
    }
async fn drain_peek_response ( & mut self ) { self . drain_feedback_msgs ( ) . await ; let mut to_send = vec ! [ ] ; let mut to_queue = vec ! [ ] ; for msg in self . queued_feedback . drain ( .. ) { if let dataflow_types :: client :: Response :: PeekResponse ( .. ) = msg { to_send . push ( msg ) ; } else { to_queue . push ( msg ) ; } } for msg in to_send { self . dataflow_client . forward_response ( msg ) ; } self . queued_feedback = to_queue ; }
Drains PeekResponse messages from the queue into coord.
Drains PeekResponse messages from the queue into coord .
&mut self
&mut self