pub async fn read_stream<'a, S: AsRef<str>>(
    connection: &GrpcClient,
    options: &ReadStreamOptions,
    stream: S,
    count: u64,
) -> crate::Result<ReadResult<BoxStream<'a, crate::Result<ResolvedEvent>>>> {
    use streams::read_req::options::stream_options::RevisionOption;
    use streams::read_req::options::{self, StreamOption, StreamOptions};
    use streams::read_req::Options;

    let read_direction = match options.direction {
        ReadDirection::Forward => 0,
        ReadDirection::Backward => 1,
    };

    let revision_option = match options.position {
        StreamPosition::Position(rev) => RevisionOption::Revision(rev),
        StreamPosition::Start => RevisionOption::Start(Empty {}),
        StreamPosition::End => RevisionOption::End(Empty {}),
    };

    let stream_identifier = Some(StreamIdentifier {
        stream_name: stream.as_ref().to_string().into_bytes(),
    });
    let stream_options = StreamOptions {
        stream_identifier,
        revision_option: Some(revision_option),
    };

    let uuid_option = options::UuidOption {
        content: Some(options::uuid_option::Content::String(Empty {})),
    };

    let credentials = options
        .credentials
        .clone()
        .or_else(|| connection.default_credentials());

    let options = Options {
        stream_option: Some(StreamOption::Stream(stream_options)),
        resolve_links: options.resolve_link_tos,
        filter_option: Some(options::FilterOption::NoFilter(Empty {})),
        count_option: Some(options::CountOption::Count(count)),
        uuid_option: Some(uuid_option),
        read_direction,
    };

    let req = streams::ReadReq {
        options: Some(options),
    };

    let mut req = Request::new(req);

    configure_auth_req(&mut req, credentials);

    connection
        .execute(|channel| async {
            let mut client = StreamsClient::new(channel.channel.clone());
            let mut stream = client.read(req).await?.into_inner();

            if let Some(resp) = stream.try_next().await? {
                match resp.content.as_ref().unwrap() {
                    streams::read_resp::Content::StreamNotFound(params) => {
                        let stream_name = std::string::String::from_utf8(
                            params
                                .stream_identifier
                                .as_ref()
                                .unwrap()
                                .stream_name
                                .clone(),
                        )
                        .expect("Don't worry this string is valid!");

                        return Ok(ReadResult::StreamNotFound(stream_name));
                    }

                    _ => {
                        let stream = stream! {
                            // We send back to the user the first event we received.
                            if let streams::read_resp::Content::Event(event) = resp.content.expect("content is defined") {
                                yield Ok(convert_proto_read_event(event));
                            }

                            loop {
                                match stream.try_next().await {
                                    Err(e) => {
                                        let e = crate::Error::from_grpc(e);

                                        channel.report_error(e.clone()).await;
                                        yield Err(e);
                                        break;
                                    }

                                    Ok(resp) => {
                                        if let Some(resp) = resp {
                                            if let streams::read_resp::Content::Event(event) = resp.content.expect("content is defined") {
                                                yield Ok(convert_proto_read_event(event));
                                            }

                                            continue;
                                        }

                                        break;
                                    }
                                }
                            }
                        };

                        let stream: BoxStream<crate::Result<ResolvedEvent>> = Box::pin(stream);

                        return Ok(ReadResult::Ok(stream));
                    }
                }
            }

            Ok(ReadResult::Ok(Box::pin(stream::empty())))
        })
        .await
}
/// Sends asynchronously the read command to the server.
Sends asynchronously the read command to the server.
Sends asynchronously the read command to the server .
pub async fn read_stream<'a, S: AsRef<str>>(
    connection: &GrpcClient,
    options: &ReadStreamOptions,
    stream: S,
    count: u64,
) -> crate::Result<ReadResult<BoxStream<'a, crate::Result<ResolvedEvent>>>> {
    use streams::read_req::options::stream_options::RevisionOption;
    use streams::read_req::options::{self, StreamOption, StreamOptions};
    use streams::read_req::Options;
    let read_direction = match options.direction {
        ReadDirection::Forward => 0,
        ReadDirection::Backward => 1,
    };
    let revision_option = match options.position {
        StreamPosition::Position(rev) => RevisionOption::Revision(rev),
        StreamPosition::Start => RevisionOption::Start(Empty {}),
        StreamPosition::End => RevisionOption::End(Empty {}),
    };
    let stream_identifier = Some(StreamIdentifier {
        stream_name: stream.as_ref().to_string().into_bytes(),
    });
    let stream_options = StreamOptions {
        stream_identifier,
        revision_option: Some(revision_option),
    };
    let uuid_option = options::UuidOption {
        content: Some(options::uuid_option::Content::String(Empty {})),
    };
    let credentials = options
        .credentials
        .clone()
        .or_else(|| connection.default_credentials());
    let options = Options {
        stream_option: Some(StreamOption::Stream(stream_options)),
        resolve_links: options.resolve_link_tos,
        filter_option: Some(options::FilterOption::NoFilter(Empty {})),
        count_option: Some(options::CountOption::Count(count)),
        uuid_option: Some(uuid_option),
        read_direction,
    };
    let req = streams::ReadReq {
        options: Some(options),
    };
    let mut req = Request::new(req);
    configure_auth_req(&mut req, credentials);
    connection
        .execute(|channel| async {
            let mut client = StreamsClient::new(channel.channel.clone());
            let mut stream = client.read(req).await?.into_inner();
            if let Some(resp) = stream.try_next().await? {
                match resp.content.as_ref().unwrap() {
                    streams::read_resp::Content::StreamNotFound(params) => {
                        let stream_name = std::string::String::from_utf8(
                            params
                                .stream_identifier
                                .as_ref()
                                .unwrap()
                                .stream_name
                                .clone(),
                        )
                        .expect("Don't worry this string is valid!");
                        return Ok(ReadResult::StreamNotFound(stream_name));
                    }
                    _ => {
                        let stream = stream! {
                            // We send back to the user the first event we received.
                            if let streams::read_resp::Content::Event(event) = resp.content.expect("content is defined") {
                                yield Ok(convert_proto_read_event(event));
                            }
                            loop {
                                match stream.try_next().await {
                                    Err(e) => {
                                        let e = crate::Error::from_grpc(e);
                                        channel.report_error(e.clone()).await;
                                        yield Err(e);
                                        break;
                                    }
                                    Ok(resp) => {
                                        if let Some(resp) = resp {
                                            if let streams::read_resp::Content::Event(event) = resp.content.expect("content is defined") {
                                                yield Ok(convert_proto_read_event(event));
                                            }
                                            continue;
                                        }
                                        break;
                                    }
                                }
                            }
                        };
                        let stream: BoxStream<crate::Result<ResolvedEvent>> = Box::pin(stream);
                        return Ok(ReadResult::Ok(stream));
                    }
                }
            }
            Ok(ReadResult::Ok(Box::pin(stream::empty())))
        })
        .await
}
pub async fn read_stream < ' a , S : AsRef < str > > ( connection : & GrpcClient , options : & ReadStreamOptions , stream : S , count : u64 , ) -> crate :: Result < ReadResult < BoxStream < ' a , crate :: Result < ResolvedEvent > > > > { use streams :: read_req :: options :: stream_options :: RevisionOption ; use streams :: read_req :: options :: { self , StreamOption , StreamOptions } ; use streams :: read_req :: Options ; let read_direction = match options . direction { ReadDirection :: Forward => 0 , ReadDirection :: Backward => 1 , } ; let revision_option = match options . position { StreamPosition :: Position ( rev ) => RevisionOption :: Revision ( rev ) , StreamPosition :: Start => RevisionOption :: Start ( Empty { } ) , StreamPosition :: End => RevisionOption :: End ( Empty { } ) , } ; let stream_identifier = Some ( StreamIdentifier { stream_name : stream . as_ref ( ) . to_string ( ) . into_bytes ( ) , } ) ; let stream_options = StreamOptions { stream_identifier , revision_option : Some ( revision_option ) , } ; let uuid_option = options :: UuidOption { content : Some ( options :: uuid_option :: Content :: String ( Empty { } ) ) , } ; let credentials = options . credentials . clone ( ) . or_else ( | | connection . default_credentials ( ) ) ; let options = Options { stream_option : Some ( StreamOption :: Stream ( stream_options ) ) , resolve_links : options . resolve_link_tos , filter_option : Some ( options :: FilterOption :: NoFilter ( Empty { } ) ) , count_option : Some ( options :: CountOption :: Count ( count ) ) , uuid_option : Some ( uuid_option ) , read_direction , } ; let req = streams :: ReadReq { options : Some ( options ) , } ; let mut req = Request :: new ( req ) ; configure_auth_req ( & mut req , credentials ) ; connection . execute ( | channel | async { let mut client = StreamsClient :: new ( channel . channel . clone ( ) ) ; let mut stream = client . read ( req ) . await ? . into_inner ( ) ; if let Some ( resp ) = stream . try_next ( ) . await ? { match resp . content . as_ref ( ) . unwrap ( ) { streams :: read_resp :: Content :: StreamNotFound ( params ) => { let stream_name = std :: string :: String :: from_utf8 ( params . stream_identifier . as_ref ( ) . unwrap ( ) . stream_name . clone ( ) , ) . expect ( " " ) ; return Ok ( ReadResult :: StreamNotFound ( stream_name ) ) ; } _ => { let stream = stream ! { // We send back to the user the first event we received. if let streams read_resp Content Event ( event ) resp content expect ( " " ) { yield Ok ( convert_proto_read_event ( event ) ) } loop { match stream try_next ( ) await { Err ( e ) { let e crate Error from_grpc ( e ) channel report_error ( e clone ( ) ) await yield Err ( e ) break } Ok ( resp ) { if let Some ( resp ) resp { if let streams read_resp Content Event ( event ) resp content expect ( " " ) { yield Ok ( convert_proto_read_event ( event ) ) } continue } break } } } } ; let stream : BoxStream < crate :: Result < ResolvedEvent > > = Box :: pin ( stream ) ; return Ok ( ReadResult :: Ok ( stream ) ) ; } } } Ok ( ReadResult :: Ok ( Box :: pin ( stream :: empty ( ) ) ) ) } ) . await }
Sends asynchronously the read command to the server.
Sends asynchronously the read command to the server .
connection
&GrpcClient
options
&ReadStreamOptions
stream
S
count
u64
connection
&GrpcClient
options
&ReadStreamOptions
stream
S
count
u64