pub async fn try_fetch<S: 'static + Select<SyncRange, Iter<SyncRecord>>>(
            keyspace: &S,
            sync_range: &SyncRange,
            retries: usize,
        ) -> anyhow::Result<SyncData> {
            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
            let _ = keyspace
                .select(sync_range)
                .consistency(Consistency::One)
                .build()?
                .send_local(ValueWorker::boxed(
                    tx,
                    keyspace.clone(),
                    sync_range.clone(),
                    retries,
                    std::marker::PhantomData,
                ));
            let select_response = rx
                .recv()
                .await
                .ok_or_else(|| anyhow::anyhow!("Expected Rx inbox to receive the sync data response"))??;
            let mut sync_data = SyncData::default();
            if let Some(mut sync_rows) = select_response {
                // Get the first row, note: the first row is always with the largest milestone_index
                let SyncRecord {
                    milestone_index,
                    logged_by,
                    ..
                } = sync_rows.next().unwrap();
                // push missing row/gap (if any)
                sync_data.process_gaps(sync_range.to, *milestone_index);
                sync_data.process_rest(&logged_by, *milestone_index, &None);
                let mut pre_ms = milestone_index;
                let mut pre_lb = logged_by;
                // Generate and identify missing gaps in order to fill them
                while let Some(SyncRecord {
                    milestone_index,
                    logged_by,
                    ..
                }) = sync_rows.next()
                {
                    // check if there are any missings
                    sync_data.process_gaps(*pre_ms, *milestone_index);
                    sync_data.process_rest(&logged_by, *milestone_index, &pre_lb);
                    pre_ms = milestone_index;
                    pre_lb = logged_by;
                }
                // pre_ms is the most recent milestone we processed
                // it's also the lowest milestone index in the select response
                // so anything < pre_ms && anything >= (self.sync_range.from - 1)
                // (lower provided sync bound) are missing
                // push missing row/gap (if any)
                sync_data.process_gaps(*pre_ms, sync_range.from - 1);
                Ok(sync_data)
            } else {
                // Everything is missing as gaps
                sync_data.process_gaps(sync_range.to, sync_range.from - 1);
                Ok(sync_data)
            }
        }
/// Try to fetch the sync data from the sync table for the provided keyspace and sync range
Try to fetch the sync data from the sync table for the provided keyspace and sync range
Try to fetch the sync data from the sync table for the provided keyspace and sync range
pub async fn try_fetch<S: 'static + Select<SyncRange, Iter<SyncRecord>>>(
            keyspace: &S,
            sync_range: &SyncRange,
            retries: usize,
        ) -> anyhow::Result<SyncData> {
            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
            let _ = keyspace
                .select(sync_range)
                .consistency(Consistency::One)
                .build()?
                .send_local(ValueWorker::boxed(
                    tx,
                    keyspace.clone(),
                    sync_range.clone(),
                    retries,
                    std::marker::PhantomData,
                ));
            let select_response = rx
                .recv()
                .await
                .ok_or_else(|| anyhow::anyhow!("Expected Rx inbox to receive the sync data response"))??;
            let mut sync_data = SyncData::default();
            if let Some(mut sync_rows) = select_response {
                // Get the first row, note: the first row is always with the largest milestone_index
                let SyncRecord {
                    milestone_index,
                    logged_by,
                    ..
                } = sync_rows.next().unwrap();
                // push missing row/gap (if any)
                sync_data.process_gaps(sync_range.to, *milestone_index);
                sync_data.process_rest(&logged_by, *milestone_index, &None);
                let mut pre_ms = milestone_index;
                let mut pre_lb = logged_by;
                // Generate and identify missing gaps in order to fill them
                while let Some(SyncRecord {
                    milestone_index,
                    logged_by,
                    ..
                }) = sync_rows.next()
                {
                    // check if there are any missings
                    sync_data.process_gaps(*pre_ms, *milestone_index);
                    sync_data.process_rest(&logged_by, *milestone_index, &pre_lb);
                    pre_ms = milestone_index;
                    pre_lb = logged_by;
                }
                // pre_ms is the most recent milestone we processed
                // it's also the lowest milestone index in the select response
                // so anything < pre_ms && anything >= (self.sync_range.from - 1)
                // (lower provided sync bound) are missing
                // push missing row/gap (if any)
                sync_data.process_gaps(*pre_ms, sync_range.from - 1);
                Ok(sync_data)
            } else {
                // Everything is missing as gaps
                sync_data.process_gaps(sync_range.to, sync_range.from - 1);
                Ok(sync_data)
            }
        }
pub async fn try_fetch < S : ' static + Select < SyncRange , Iter < SyncRecord > > > ( keyspace : & S , sync_range : & SyncRange , retries : usize , ) -> anyhow :: Result < SyncData > { let ( tx , mut rx ) = tokio :: sync :: mpsc :: unbounded_channel ( ) ; let _ = keyspace . select ( sync_range ) . consistency ( Consistency :: One ) . build ( ) ? . send_local ( ValueWorker :: boxed ( tx , keyspace . clone ( ) , sync_range . clone ( ) , retries , std :: marker :: PhantomData , ) ) ; let select_response = rx . recv ( ) . await . ok_or_else ( | | anyhow :: anyhow ! ( " " ) ) ? ? ; let mut sync_data = SyncData :: default ( ) ; if let Some ( mut sync_rows ) = select_response { // Get the first row, note: the first row is always with the largest milestone_index let SyncRecord { milestone_index , logged_by , .. } = sync_rows . next ( ) . unwrap ( ) ; // push missing row/gap (if any) sync_data . process_gaps ( sync_range . to , * milestone_index ) ; sync_data . process_rest ( & logged_by , * milestone_index , & None ) ; let mut pre_ms = milestone_index ; let mut pre_lb = logged_by ; // Generate and identify missing gaps in order to fill them while let Some ( SyncRecord { milestone_index , logged_by , .. } ) = sync_rows . next ( ) { // check if there are any missings sync_data . process_gaps ( * pre_ms , * milestone_index ) ; sync_data . process_rest ( & logged_by , * milestone_index , & pre_lb ) ; pre_ms = milestone_index ; pre_lb = logged_by ; } // pre_ms is the most recent milestone we processed // it's also the lowest milestone index in the select response // so anything < pre_ms && anything >= (self.sync_range.from - 1) // (lower provided sync bound) are missing // push missing row/gap (if any) sync_data . process_gaps ( * pre_ms , sync_range . from - 1 ) ; Ok ( sync_data ) } else { // Everything is missing as gaps sync_data . process_gaps ( sync_range . to , sync_range . from - 1 ) ; Ok ( sync_data ) } }
Try to fetch the sync data from the sync table for the provided keyspace and sync range
Try to fetch the sync data from the sync table for the provided keyspace and sync range
keyspace
&S
sync_range
&SyncRange
retries
usize