pub async fn receive_with_options(
        &mut self,
        opts: ReceiveOptions,
    ) -> Result<Option<Message>, Error> {
        loop {
            if let Some(handle) = self.buffer.pop_front() {
                let message = handle.message.unwrap();
                let timestamp = message.publish_time.unwrap();
                let message = Message {
                    client: self.client.clone(),
                    subscription_name: self.name.clone(),
                    data: message.data,
                    message_id: message.message_id,
                    ack_id: handle.ack_id,
                    attributes: message.attributes,
                    publish_time: chrono::NaiveDateTime::from_timestamp(
                        timestamp.seconds,
                        timestamp.nanos as u32,
                    ),
                };
                break Ok(Some(message));
            } else {
                match self.pull(&opts).await {
                    Ok(messages) => {
                        if messages.is_empty() && opts.return_immediately {
                            break Ok(None);
                        } else {
                            self.buffer.extend(messages);
                        }
                    }
                    Err(_) => {}
                }
            }
        }
    }
/// Receive the next message from the subscription with options.
Receive the next message from the subscription with options.
Receive the next message from the subscription with options .
pub async fn receive_with_options(
        &mut self,
        opts: ReceiveOptions,
    ) -> Result<Option<Message>, Error> {
        loop {
            if let Some(handle) = self.buffer.pop_front() {
                let message = handle.message.unwrap();
                let timestamp = message.publish_time.unwrap();
                let message = Message {
                    client: self.client.clone(),
                    subscription_name: self.name.clone(),
                    data: message.data,
                    message_id: message.message_id,
                    ack_id: handle.ack_id,
                    attributes: message.attributes,
                    publish_time: chrono::NaiveDateTime::from_timestamp(
                        timestamp.seconds,
                        timestamp.nanos as u32,
                    ),
                };
                break Ok(Some(message));
            } else {
                match self.pull(&opts).await {
                    Ok(messages) => {
                        if messages.is_empty() && opts.return_immediately {
                            break Ok(None);
                        } else {
                            self.buffer.extend(messages);
                        }
                    }
                    Err(_) => {}
                }
            }
        }
    }
pub async fn receive_with_options ( & mut self , opts : ReceiveOptions , ) -> Result < Option < Message > , Error > { loop { if let Some ( handle ) = self . buffer . pop_front ( ) { let message = handle . message . unwrap ( ) ; let timestamp = message . publish_time . unwrap ( ) ; let message = Message { client : self . client . clone ( ) , subscription_name : self . name . clone ( ) , data : message . data , message_id : message . message_id , ack_id : handle . ack_id , attributes : message . attributes , publish_time : chrono :: NaiveDateTime :: from_timestamp ( timestamp . seconds , timestamp . nanos as u32 , ) , } ; break Ok ( Some ( message ) ) ; } else { match self . pull ( & opts ) . await { Ok ( messages ) => { if messages . is_empty ( ) && opts . return_immediately { break Ok ( None ) ; } else { self . buffer . extend ( messages ) ; } } Err ( _ ) => { } } } } }
Receive the next message from the subscription with options.
Receive the next message from the subscription with options .
&mut self
opts
ReceiveOptions
&mut self
opts
ReceiveOptions