pub async fn try_fetch<S: 'static + Select<SyncRange, Iter<AnalyticRecord>>>(
            keyspace: &S,
            sync_range: &SyncRange,
            retries: usize,
            page_size: i32,
        ) -> anyhow::Result<AnalyticsData> {
            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
            Self::query_analytics_table(keyspace, sync_range, retries, tx.clone(), page_size, None)?;
            let mut analytics_data = AnalyticsData::default();
            while let Some(mut records) = rx
                .recv()
                .await
                .ok_or_else(|| anyhow::anyhow!("Unable to fetch the analytics response"))??
            {
                if records.has_more_pages() {
                    // request next page
                    let paging_state = records.take_paging_state();
                    // this will request the next page, and value worker will pass it to us through rx
                    Self::query_analytics_table(keyspace, sync_range, retries, tx.clone(), page_size, paging_state)?;
                    // Gets the first record in the page result, which is used to trigger accumulation
                    analytics_data.try_trigger(&mut records).await;
                } else {
                    // no more pages to fetch, therefore we process whatever we received, ..
                    analytics_data.try_trigger(&mut records).await;
                    // break the while
                    break;
                }
            }
            Ok(analytics_data)
        }
/// Try to fetch the analytics data from the analytics table for the provided keyspace and sync range
Try to fetch the analytics data from the analytics table for the provided keyspace and sync range
Try to fetch the analytics data from the analytics table for the provided keyspace and sync range
pub async fn try_fetch<S: 'static + Select<SyncRange, Iter<AnalyticRecord>>>(
            keyspace: &S,
            sync_range: &SyncRange,
            retries: usize,
            page_size: i32,
        ) -> anyhow::Result<AnalyticsData> {
            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
            Self::query_analytics_table(keyspace, sync_range, retries, tx.clone(), page_size, None)?;
            let mut analytics_data = AnalyticsData::default();
            while let Some(mut records) = rx
                .recv()
                .await
                .ok_or_else(|| anyhow::anyhow!("Unable to fetch the analytics response"))??
            {
                if records.has_more_pages() {
                    // request next page
                    let paging_state = records.take_paging_state();
                    // this will request the next page, and value worker will pass it to us through rx
                    Self::query_analytics_table(keyspace, sync_range, retries, tx.clone(), page_size, paging_state)?;
                    // Gets the first record in the page result, which is used to trigger accumulation
                    analytics_data.try_trigger(&mut records).await;
                } else {
                    // no more pages to fetch, therefore we process whatever we received, ..
                    analytics_data.try_trigger(&mut records).await;
                    // break the while
                    break;
                }
            }
            Ok(analytics_data)
        }
pub async fn try_fetch < S : ' static + Select < SyncRange , Iter < AnalyticRecord > > > ( keyspace : & S , sync_range : & SyncRange , retries : usize , page_size : i32 , ) -> anyhow :: Result < AnalyticsData > { let ( tx , mut rx ) = tokio :: sync :: mpsc :: unbounded_channel ( ) ; Self :: query_analytics_table ( keyspace , sync_range , retries , tx . clone ( ) , page_size , None ) ? ; let mut analytics_data = AnalyticsData :: default ( ) ; while let Some ( mut records ) = rx . recv ( ) . await . ok_or_else ( | | anyhow :: anyhow ! ( " " ) ) ? ? { if records . has_more_pages ( ) { // request next page let paging_state = records . take_paging_state ( ) ; // this will request the next page, and value worker will pass it to us through rx Self :: query_analytics_table ( keyspace , sync_range , retries , tx . clone ( ) , page_size , paging_state ) ? ; // Gets the first record in the page result, which is used to trigger accumulation analytics_data . try_trigger ( & mut records ) . await ; } else { // no more pages to fetch, therefore we process whatever we received, .. analytics_data . try_trigger ( & mut records ) . await ; // break the while break ; } } Ok ( analytics_data ) }
Try to fetch the analytics data from the analytics table for the provided keyspace and sync range
Try to fetch the analytics data from the analytics table for the provided keyspace and sync range
keyspace
&S
sync_range
&SyncRange
retries
usize
page_size
i32