pub async fn subscribe_to_stream<'a, StreamName>(
        &self,
        stream_name: StreamName,
        options: &SubscribeToStreamOptions,
    ) -> crate::Result<BoxStream<'a, crate::Result<SubEvent>>>
    where
        StreamName: AsRef<str>,
    {
        match options.retry.as_ref().cloned() {
            None => commands::subscribe_to_stream(&self.client, stream_name, options).await,
            Some(retry) => {
                let stream_name = stream_name.as_ref().to_string();
                let mut attempt_count = 1usize;
                let mut offset = options.position;
                let client = self.client.clone();
                let mut options = options.clone();
                let result = async_stream::stream! {
                    loop {
                        match commands::subscribe_to_stream(&client, stream_name.as_str(), &options).await {
                            Err(e) => {
                                if attempt_count == retry.limit {
                                    error!("Subscription: maximum retry threshold reached, cause: {}", e);

                                    yield Err(e);
                                    break;
                                }

                                error!("Subscription: attempt ({}/{}) failure, cause: {}", attempt_count, retry.limit, e);
                                attempt_count += 1;
                                tokio::time::sleep(retry.delay).await;
                            }
                            Ok(mut stream) => {
                                loop {
                                    match stream.try_next().await {
                                        Ok(sub_event) => {
                                            let sub_event = sub_event.expect("to be defined");
                                            match sub_event {
                                                crate::types::SubEvent::EventAppeared(event) => {
                                                    offset = crate::types::StreamPosition::Position(event.get_original_event().revision);
                                                    yield Ok(crate::types::SubEvent::EventAppeared(event));
                                                }

                                                ignored => yield Ok(ignored),
                                            }
                                        }
                                        Err(e) => {
                                            attempt_count = 1;
                                            options = options.position(offset);

                                            error!("Subscription dropped cause: {}. Reconnecting", e);
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                };

                let result: BoxStream<crate::Result<SubEvent>> = Box::pin(result);

                Ok(result)
            }
        }
    }
/// Subscribes to a given stream. This kind of subscription specifies a
/// starting point (by default, the beginning of a stream). For a regular
/// stream, that starting point will be an event number. For the system
/// stream `$all`, it will be a position in the transaction file
/// (see [`subscribe_to_all`]). This subscription will fetch every event
/// until the end of the stream, then will dispatch subsequently written
/// events.
///
/// For example, if a starting point of 50 is specified when a stream has
/// 100 events in it, the subscriber can expect to see events 51 through
/// 100, and then any events subsequently written events until such time
/// as the subscription is dropped or closed.
///
/// [`subscribe_to_all`]: #method.subscribe_to_all_from
Subscribes to a given stream. This kind of subscription specifies a
starting point (by default, the beginning of a stream). For a regular
stream, that starting point will be an event number. For the system
stream `$all`, it will be a position in the transaction file
. This subscription will fetch every event
until the end of the stream, then will dispatch subsequently written
events.

For example, if a starting point of 50 is specified when a stream has
100 events in it, the subscriber can expect to see events 51 through
100, and then any events subsequently written events until such time
as the subscription is dropped or closed.


Subscribes to a given stream . This kind of subscription specifies a starting point ( by default the beginning of a stream ) . For a regular stream that starting point will be an event number . For the system stream ` $all ` it will be a position in the transaction file . This subscription will fetch every event until the end of the stream then will dispatch subsequently written events . For example if a starting point of 50 is specified when a stream has 100 events in it the subscriber can expect to see events 51 through 100 and then any events subsequently written events until such time as the subscription is dropped or closed .
pub async fn subscribe_to_stream<'a, StreamName>(
        &self,
        stream_name: StreamName,
        options: &SubscribeToStreamOptions,
    ) -> crate::Result<BoxStream<'a, crate::Result<SubEvent>>>
    where
        StreamName: AsRef<str>,
    {
        match options.retry.as_ref().cloned() {
            None => commands::subscribe_to_stream(&self.client, stream_name, options).await,
            Some(retry) => {
                let stream_name = stream_name.as_ref().to_string();
                let mut attempt_count = 1usize;
                let mut offset = options.position;
                let client = self.client.clone();
                let mut options = options.clone();
                let result = async_stream::stream! {
                    loop {
                        match commands::subscribe_to_stream(&client, stream_name.as_str(), &options).await {
                            Err(e) => {
                                if attempt_count == retry.limit {
                                    error!("Subscription: maximum retry threshold reached, cause: {}", e);
                                    yield Err(e);
                                    break;
                                }
                                error!("Subscription: attempt ({}/{}) failure, cause: {}", attempt_count, retry.limit, e);
                                attempt_count += 1;
                                tokio::time::sleep(retry.delay).await;
                            }
                            Ok(mut stream) => {
                                loop {
                                    match stream.try_next().await {
                                        Ok(sub_event) => {
                                            let sub_event = sub_event.expect("to be defined");
                                            match sub_event {
                                                crate::types::SubEvent::EventAppeared(event) => {
                                                    offset = crate::types::StreamPosition::Position(event.get_original_event().revision);
                                                    yield Ok(crate::types::SubEvent::EventAppeared(event));
                                                }
                                                ignored => yield Ok(ignored),
                                            }
                                        }
                                        Err(e) => {
                                            attempt_count = 1;
                                            options = options.position(offset);
                                            error!("Subscription dropped cause: {}. Reconnecting", e);
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                };
                let result: BoxStream<crate::Result<SubEvent>> = Box::pin(result);
                Ok(result)
            }
        }
    }
pub async fn subscribe_to_stream < ' a , StreamName > ( & self , stream_name : StreamName , options : & SubscribeToStreamOptions , ) -> crate :: Result < BoxStream < ' a , crate :: Result < SubEvent > > > where StreamName : AsRef < str > , { match options . retry . as_ref ( ) . cloned ( ) { None => commands :: subscribe_to_stream ( & self . client , stream_name , options ) . await , Some ( retry ) => { let stream_name = stream_name . as_ref ( ) . to_string ( ) ; let mut attempt_count = 1usize ; let mut offset = options . position ; let client = self . client . clone ( ) ; let mut options = options . clone ( ) ; let result = async_stream :: stream ! { loop { match commands subscribe_to_stream ( client stream_name as_str ( ) options ) await { Err ( e ) { if attempt_count retry limit { error ( " " e ) yield Err ( e ) break } error ( " " attempt_count retry limit e ) attempt_count 1 tokio time sleep ( retry delay ) await } Ok ( mut stream ) { loop { match stream try_next ( ) await { Ok ( sub_event ) { let sub_event sub_event expect ( " " ) match sub_event { crate types SubEvent EventAppeared ( event ) { offset crate types StreamPosition Position ( event get_original_event ( ) revision ) yield Ok ( crate types SubEvent EventAppeared ( event ) ) } ignored yield Ok ( ignored ) } } Err ( e ) { attempt_count 1 options options position ( offset ) error ( " " e ) break } } } } } } } ; let result : BoxStream < crate :: Result < SubEvent > > = Box :: pin ( result ) ; Ok ( result ) } } }
Subscribes to a given stream.
Subscribes to a given stream .
&self
stream_name
StreamName
options
&SubscribeToStreamOptions
&self
stream_name
StreamName
options
&SubscribeToStreamOptions