async fn reduce(
    mut link_status_observer: Observer<LinkStatusMap>,
    observable: Observable<BTreeMap<NodeId, LinkMetrics>>,
) {
    while let Some(link_status) = link_status_observer.next().await {
        let mut new_status: BTreeMap<NodeId, LinkMetrics> = Default::default();
        for (&node_link_id, &(node_id, round_trip_time)) in link_status.iter() {
            let metrics = LinkMetrics { node_link_id, round_trip_time };
            new_status
                .entry(node_id)
                .and_modify(|link_metrics| {
                    if metrics.score() > link_metrics.score() {
                        *link_metrics = metrics.clone();
                    }
                })
                .or_insert(metrics);
        }
        observable.push(new_status).await;
    }
}
/// Continually condenses a map of the status of all links into a map giving the status of the best
/// available link by which to reach each node.
Continually condenses a map of the status of all links into a map giving the status of the best
available link by which to reach each node.
Continually condenses a map of the status of all links into a map giving the status of the best available link by which to reach each node .
async fn reduce(
    mut link_status_observer: Observer<LinkStatusMap>,
    observable: Observable<BTreeMap<NodeId, LinkMetrics>>,
) {
    while let Some(link_status) = link_status_observer.next().await {
        let mut new_status: BTreeMap<NodeId, LinkMetrics> = Default::default();
        for (&node_link_id, &(node_id, round_trip_time)) in link_status.iter() {
            let metrics = LinkMetrics { node_link_id, round_trip_time };
            new_status
                .entry(node_id)
                .and_modify(|link_metrics| {
                    if metrics.score() > link_metrics.score() {
                        *link_metrics = metrics.clone();
                    }
                })
                .or_insert(metrics);
        }
        observable.push(new_status).await;
    }
}
async fn reduce ( mut link_status_observer : Observer < LinkStatusMap > , observable : Observable < BTreeMap < NodeId , LinkMetrics > > , ) { while let Some ( link_status ) = link_status_observer . next ( ) . await { let mut new_status : BTreeMap < NodeId , LinkMetrics > = Default :: default ( ) ; for ( & node_link_id , & ( node_id , round_trip_time ) ) in link_status . iter ( ) { let metrics = LinkMetrics { node_link_id , round_trip_time } ; new_status . entry ( node_id ) . and_modify ( | link_metrics | { if metrics . score ( ) > link_metrics . score ( ) { * link_metrics = metrics . clone ( ) ; } } ) . or_insert ( metrics ) ; } observable . push ( new_status ) . await ; } }
Continually condenses a map of the status of all links into a map giving the status of the best
available link by which to reach each node.
Continually condenses a map of the status of all links into a map giving the status of the best available link by which to reach each node .
link_status_observer
Observer<LinkStatusMap>
observable
Observable<BTreeMap<NodeId, LinkMetrics>>
link_status_observer
Observer<LinkStatusMap>
observable
Observable<BTreeMap<NodeId, LinkMetrics>>