pub async fn collect<T: DeserializeOwned>(
        &mut self,
        timeout: Duration,
        reason: &str,
    ) -> Vec<T> {
        let start = Instant::now();
        let mut items = Vec::new();
        loop {
            let elapsed = start.elapsed();
            if elapsed >= timeout {
                break;
            }
            // keep collecting while there is time remaining
            match self.next(timeout - elapsed).await {
                SubscriptionNextResult::Item(item) => items.push(item),
                SubscriptionNextResult::Cancelled | SubscriptionNextResult::Timeout => break,
                SubscriptionNextResult::Err(s) => {
                    // log corrupt messages but continue receiving until timeout
                    error!("corrupt message received {}: {}", reason, s);
                }
            }
        }
        items
    }
/// Collect results until timeout has elapsed
Collect results until timeout has elapsed
Collect results until timeout has elapsed
pub async fn collect<T: DeserializeOwned>(
        &mut self,
        timeout: Duration,
        reason: &str,
    ) -> Vec<T> {
        let start = Instant::now();
        let mut items = Vec::new();
        loop {
            let elapsed = start.elapsed();
            if elapsed >= timeout {
                break;
            }
            // keep collecting while there is time remaining
            match self.next(timeout - elapsed).await {
                SubscriptionNextResult::Item(item) => items.push(item),
                SubscriptionNextResult::Cancelled | SubscriptionNextResult::Timeout => break,
                SubscriptionNextResult::Err(s) => {
                    // log corrupt messages but continue receiving until timeout
                    error!("corrupt message received {}: {}", reason, s);
                }
            }
        }
        items
    }
pub async fn collect < T : DeserializeOwned > ( & mut self , timeout : Duration , reason : & str , ) -> Vec < T > { let start = Instant :: now ( ) ; let mut items = Vec :: new ( ) ; loop { let elapsed = start . elapsed ( ) ; if elapsed >= timeout { break ; } // keep collecting while there is time remaining match self . next ( timeout - elapsed ) . await { SubscriptionNextResult :: Item ( item ) => items . push ( item ) , SubscriptionNextResult :: Cancelled | SubscriptionNextResult :: Timeout => break , SubscriptionNextResult :: Err ( s ) => { // log corrupt messages but continue receiving until timeout error ! ( " " reason s ) ; } } } items }
Collect results until timeout has elapsed
Collect results until timeout has elapsed
&mut self
timeout
Duration
reason
&str
&mut self
timeout
Duration
reason
&str