pub fn absorb_response(&mut self, shard_id: usize, message: Response) -> Option<Response> {
            match message {
                Response::FrontierUppers(mut list) => {
                    // Fold updates into the maintained antichain, and report
                    // any net changes to the minimal antichain itself.
                    let mut reactions = ChangeBatch::new();
                    for (id, changes) in list.iter_mut() {
                        // We may receive changes to identifiers that are no longer tracked;
                        // do not worry about them in that case (a benign race condition).
                        if let Some(frontier) = self.uppers.get_mut(id) {
                            for (time, diff) in frontier.update_iter(changes.drain()) {
                                reactions.update(time, diff);
                            }
                            std::mem::swap(changes, &mut reactions);
                        } else {
                            changes.clear();
                        }
                    }
                    // TODO: The following line would be great, but is not permitted by `list.retain()`.
                    // list.retain_mut(|(_, changes)| !changes.is_empty());
                    if !list.is_empty() {
                        Some(Response::FrontierUppers(list))
                    } else {
                        None
                    }
                }
                Response::PeekResponse(connection, response) => {
                    // Incorporate new peek responses; awaiting all responses.
                    let entry = self
                        .peek_responses
                        .entry(connection)
                        .or_insert_with(Default::default);
                    let novel = entry.insert(shard_id, response);
                    assert!(novel.is_none(), "Duplicate peek response");
                    // We may be ready to respond.
                    if entry.len() == self.parts {
                        let mut response = PeekResponse::Rows(Vec::new());
                        for (_part, r) in std::mem::take(entry).into_iter() {
                            response = match (response, r) {
                                (_, PeekResponse::Canceled) => PeekResponse::Canceled,
                                (PeekResponse::Canceled, _) => PeekResponse::Canceled,
                                (_, PeekResponse::Error(e)) => PeekResponse::Error(e),
                                (PeekResponse::Error(e), _) => PeekResponse::Error(e),
                                (PeekResponse::Rows(mut rows), PeekResponse::Rows(r)) => {
                                    rows.extend(r.into_iter());
                                    PeekResponse::Rows(rows)
                                }
                            };
                        }
                        self.peek_responses.remove(&connection);
                        Some(Response::PeekResponse(connection, response))
                    } else {
                        None
                    }
                }
                message => {
                    // TimestampBindings and TailResponses are mirrored out,
                    // as they do not seem to contain worker-specific information.
                    Some(message)
                }
            }
        }
/// Absorbs a response, and produces response that should be emitted.
Absorbs a response, and produces response that should be emitted.
Absorbs a response and produces response that should be emitted .
pub fn absorb_response(&mut self, shard_id: usize, message: Response) -> Option<Response> {
            match message {
                Response::FrontierUppers(mut list) => {
                    // Fold updates into the maintained antichain, and report
                    // any net changes to the minimal antichain itself.
                    let mut reactions = ChangeBatch::new();
                    for (id, changes) in list.iter_mut() {
                        // We may receive changes to identifiers that are no longer tracked;
                        // do not worry about them in that case (a benign race condition).
                        if let Some(frontier) = self.uppers.get_mut(id) {
                            for (time, diff) in frontier.update_iter(changes.drain()) {
                                reactions.update(time, diff);
                            }
                            std::mem::swap(changes, &mut reactions);
                        } else {
                            changes.clear();
                        }
                    }
                    // TODO: The following line would be great, but is not permitted by `list.retain()`.
                    // list.retain_mut(|(_, changes)| !changes.is_empty());
                    if !list.is_empty() {
                        Some(Response::FrontierUppers(list))
                    } else {
                        None
                    }
                }
                Response::PeekResponse(connection, response) => {
                    // Incorporate new peek responses; awaiting all responses.
                    let entry = self
                        .peek_responses
                        .entry(connection)
                        .or_insert_with(Default::default);
                    let novel = entry.insert(shard_id, response);
                    assert!(novel.is_none(), "Duplicate peek response");
                    // We may be ready to respond.
                    if entry.len() == self.parts {
                        let mut response = PeekResponse::Rows(Vec::new());
                        for (_part, r) in std::mem::take(entry).into_iter() {
                            response = match (response, r) {
                                (_, PeekResponse::Canceled) => PeekResponse::Canceled,
                                (PeekResponse::Canceled, _) => PeekResponse::Canceled,
                                (_, PeekResponse::Error(e)) => PeekResponse::Error(e),
                                (PeekResponse::Error(e), _) => PeekResponse::Error(e),
                                (PeekResponse::Rows(mut rows), PeekResponse::Rows(r)) => {
                                    rows.extend(r.into_iter());
                                    PeekResponse::Rows(rows)
                                }
                            };
                        }
                        self.peek_responses.remove(&connection);
                        Some(Response::PeekResponse(connection, response))
                    } else {
                        None
                    }
                }
                message => {
                    // TimestampBindings and TailResponses are mirrored out,
                    // as they do not seem to contain worker-specific information.
                    Some(message)
                }
            }
        }
pub fn absorb_response ( & mut self , shard_id : usize , message : Response ) -> Option < Response > { match message { Response :: FrontierUppers ( mut list ) => { // Fold updates into the maintained antichain, and report // any net changes to the minimal antichain itself. let mut reactions = ChangeBatch :: new ( ) ; for ( id , changes ) in list . iter_mut ( ) { // We may receive changes to identifiers that are no longer tracked; // do not worry about them in that case (a benign race condition). if let Some ( frontier ) = self . uppers . get_mut ( id ) { for ( time , diff ) in frontier . update_iter ( changes . drain ( ) ) { reactions . update ( time , diff ) ; } std :: mem :: swap ( changes , & mut reactions ) ; } else { changes . clear ( ) ; } } // TODO: The following line would be great, but is not permitted by `list.retain()`. // list.retain_mut(|(_, changes)| !changes.is_empty()); if ! list . is_empty ( ) { Some ( Response :: FrontierUppers ( list ) ) } else { None } } Response :: PeekResponse ( connection , response ) => { // Incorporate new peek responses; awaiting all responses. let entry = self . peek_responses . entry ( connection ) . or_insert_with ( Default :: default ) ; let novel = entry . insert ( shard_id , response ) ; assert ! ( novel is_none ( ) " " ) ; // We may be ready to respond. if entry . len ( ) == self . parts { let mut response = PeekResponse :: Rows ( Vec :: new ( ) ) ; for ( _part , r ) in std :: mem :: take ( entry ) . into_iter ( ) { response = match ( response , r ) { ( _ , PeekResponse :: Canceled ) => PeekResponse :: Canceled , ( PeekResponse :: Canceled , _ ) => PeekResponse :: Canceled , ( _ , PeekResponse :: Error ( e ) ) => PeekResponse :: Error ( e ) , ( PeekResponse :: Error ( e ) , _ ) => PeekResponse :: Error ( e ) , ( PeekResponse :: Rows ( mut rows ) , PeekResponse :: Rows ( r ) ) => { rows . extend ( r . into_iter ( ) ) ; PeekResponse :: Rows ( rows ) } } ; } self . peek_responses . remove ( & connection ) ; Some ( Response :: PeekResponse ( connection , response ) ) } else { None } } message => { // TimestampBindings and TailResponses are mirrored out, // as they do not seem to contain worker-specific information. Some ( message ) } } }
Absorbs a response, and produces response that should be emitted.
Absorbs a response and produces response that should be emitted .
&mut self
shard_id
usize
message
Response
&mut self
shard_id
usize
message
Response