/// Start a stream of incoming messages with options.
Start a stream of incoming messages with options.
Start a stream of incoming messages with options .
pub async fn stream_with_options(
        &mut self,
        opts: StreamingOptions,
    ) -> Result<impl Stream<Item = Result<Message, Status>>, Error> {
        let filter_redeliveries = opts.filter_redeliveries;
        let (streaming, tx) = self.pull_streaming(opts).await?;
        let client = self.client.clone();
        let name = self.name.clone();
        let tx = Arc::new(Mutex::new(tx));
        Ok(futures::stream::unfold(streaming, |mut res| async {
            match res.message().await {
                Ok(Some(v)) => Some((Ok(v), res)),
                Ok(None) => None,
                // TODO: Better error handling?
                Err(err) => Some((Err(err), res)),
            }
        })
        .then({
            let tx = tx.clone();
            move |res| {
                let tx = tx.clone();
                async move {
                    match res {
                        Ok(v) => {
                            // TODO: Better end-user message acknowledgement mechanism
                            let mut tx = tx.lock().await;
                            Ok(tx
                                .send(
                                    v.received_messages
                                        .iter()
                                        .map(|m| m.ack_id.clone())
                                        .collect(),
                                )
                                .map(|res| {
                                    res.map(|()| v.received_messages).expect("Received closed")
                                })
                                .await)
                        }
                        Err(err) => Err(err),
                    }
                }
            }
        })
        .flat_map(|v: Result<Vec<ReceivedMessage>, Status>| match v {
            Ok(v) => futures::stream::iter(v).map(Ok).boxed(),
            Err(err) => futures::stream::once(futures::future::ready(Err(err))).boxed(),
        })
        .filter(move |m| {
            futures::future::ready(if filter_redeliveries {
                m.as_ref().map(|m| m.delivery_attempt == 0).unwrap_or(true) // Propagate errors through
            } else {
                true
            })
        })
        .filter(|m| futures::future::ready(m.as_ref().map(|m| m.message.is_some()).unwrap_or(true))) // Propagate errors through
        .map(move |m: Result<ReceivedMessage, Status>| {
            m.map(|m| {
                let inner_msg = m.message.unwrap();
                let raw_publish_time = inner_msg.publish_time.unwrap_or_default();
                Message {
                    client: client.clone(),
                    subscription_name: name.clone(),
                    ack_id: m.ack_id,
                    message_id: inner_msg.message_id,
                    publish_time: NaiveDateTime::from_timestamp(
                        raw_publish_time.seconds,
                        raw_publish_time.nanos as u32,
                    ),
                    attributes: inner_msg.attributes,
                    data: inner_msg.data,
                }
            })
        }))
    }
pub async fn stream_with_options ( & mut self , opts : StreamingOptions , ) -> Result < impl Stream < Item = Result < Message , Status > > , Error > { let filter_redeliveries = opts . filter_redeliveries ; let ( streaming , tx ) = self . pull_streaming ( opts ) . await ? ; let client = self . client . clone ( ) ; let name = self . name . clone ( ) ; let tx = Arc :: new ( Mutex :: new ( tx ) ) ; Ok ( futures :: stream :: unfold ( streaming , | mut res | async { match res . message ( ) . await { Ok ( Some ( v ) ) => Some ( ( Ok ( v ) , res ) ) , Ok ( None ) => None , // TODO: Better error handling? Err ( err ) => Some ( ( Err ( err ) , res ) ) , } } ) . then ( { let tx = tx . clone ( ) ; move | res | { let tx = tx . clone ( ) ; async move { match res { Ok ( v ) => { // TODO: Better end-user message acknowledgement mechanism let mut tx = tx . lock ( ) . await ; Ok ( tx . send ( v . received_messages . iter ( ) . map ( | m | m . ack_id . clone ( ) ) . collect ( ) , ) . map ( | res | { res . map ( | ( ) | v . received_messages ) . expect ( " " ) } ) . await ) } Err ( err ) => Err ( err ) , } } } } ) . flat_map ( | v : Result < Vec < ReceivedMessage > , Status > | match v { Ok ( v ) => futures :: stream :: iter ( v ) . map ( Ok ) . boxed ( ) , Err ( err ) => futures :: stream :: once ( futures :: future :: ready ( Err ( err ) ) ) . boxed ( ) , } ) . filter ( move | m | { futures :: future :: ready ( if filter_redeliveries { m . as_ref ( ) . map ( | m | m . delivery_attempt == 0 ) . unwrap_or ( true ) // Propagate errors through } else { true } ) } ) . filter ( | m | futures :: future :: ready ( m . as_ref ( ) . map ( | m | m . message . is_some ( ) ) . unwrap_or ( true ) ) ) // Propagate errors through . map ( move | m : Result < ReceivedMessage , Status > | { m . map ( | m | { let inner_msg = m . message . unwrap ( ) ; let raw_publish_time = inner_msg . publish_time . unwrap_or_default ( ) ; Message { client : client . clone ( ) , subscription_name : name . clone ( ) , ack_id : m . ack_id , message_id : inner_msg . message_id , publish_time : NaiveDateTime :: from_timestamp ( raw_publish_time . seconds , raw_publish_time . nanos as u32 , ) , attributes : inner_msg . attributes , data : inner_msg . data , } } ) } ) ) }
Start a stream of incoming messages with options.
Start a stream of incoming messages with options .
&mut self
opts
StreamingOptions