pub async fn load_prefix_futures<'a, P>(
    prefix: P,
    config: RecordIndexerConfig,
) -> Result<impl Stream<Item = impl Future<Output = Result<impl Stream<Item = Result<RecordIndex>>>>>>
where
    P: Into<Cow<'a, str>>,
{
    let (dir, file_name_prefix) = utils::split_prefix(prefix);
    let dir = Path::new(&dir);
    let file_name_prefix = Arc::new(file_name_prefix);

    // filter paths
    let mut paths: Vec<_> = dir
        .read_dir()
        .await?
        .map(|result| result.map_err(Error::from))
        .try_filter_map(|entry| {
            let file_name_prefix = file_name_prefix.clone();

            async move {
                if !entry.metadata().await?.is_file() {
                    return Ok(None);
                }
                let file_name = PathBuf::from(entry.file_name());
                let path = file_name
                    .starts_with(&*file_name_prefix)
                    .then(|| std::path::PathBuf::from(entry.path().into_os_string()));
                Ok(path)
            }
        })
        .try_collect()
        .await?;

    // sort paths
    // TODO: fix blocking?
    paths.sort();

    // construct dataset
    let stream = load_paths_futures(paths, config);
    Ok(stream)
}
/// Generate futures that load record indexes from files specified by a prefix.
Generate futures that load record indexes from files specified by a prefix.
Generate futures that load record indexes from files specified by a prefix .
pub async fn load_prefix_futures<'a, P>(
    prefix: P,
    config: RecordIndexerConfig,
) -> Result<impl Stream<Item = impl Future<Output = Result<impl Stream<Item = Result<RecordIndex>>>>>>
where
    P: Into<Cow<'a, str>>,
{
    let (dir, file_name_prefix) = utils::split_prefix(prefix);
    let dir = Path::new(&dir);
    let file_name_prefix = Arc::new(file_name_prefix);
    // filter paths
    let mut paths: Vec<_> = dir
        .read_dir()
        .await?
        .map(|result| result.map_err(Error::from))
        .try_filter_map(|entry| {
            let file_name_prefix = file_name_prefix.clone();
            async move {
                if !entry.metadata().await?.is_file() {
                    return Ok(None);
                }
                let file_name = PathBuf::from(entry.file_name());
                let path = file_name
                    .starts_with(&*file_name_prefix)
                    .then(|| std::path::PathBuf::from(entry.path().into_os_string()));
                Ok(path)
            }
        })
        .try_collect()
        .await?;
    // sort paths
    // TODO: fix blocking?
    paths.sort();
    // construct dataset
    let stream = load_paths_futures(paths, config);
    Ok(stream)
}
pub async fn load_prefix_futures < ' a , P > ( prefix : P , config : RecordIndexerConfig , ) -> Result < impl Stream < Item = impl Future < Output = Result < impl Stream < Item = Result < RecordIndex > > > > > > where P : Into < Cow < ' a , str > > , { let ( dir , file_name_prefix ) = utils :: split_prefix ( prefix ) ; let dir = Path :: new ( & dir ) ; let file_name_prefix = Arc :: new ( file_name_prefix ) ; // filter paths let mut paths : Vec < _ > = dir . read_dir ( ) . await ? . map ( | result | result . map_err ( Error :: from ) ) . try_filter_map ( | entry | { let file_name_prefix = file_name_prefix . clone ( ) ; async move { if ! entry . metadata ( ) . await ? . is_file ( ) { return Ok ( None ) ; } let file_name = PathBuf :: from ( entry . file_name ( ) ) ; let path = file_name . starts_with ( & * file_name_prefix ) . then ( | | std :: path :: PathBuf :: from ( entry . path ( ) . into_os_string ( ) ) ) ; Ok ( path ) } } ) . try_collect ( ) . await ? ; // sort paths // TODO: fix blocking? paths . sort ( ) ; // construct dataset let stream = load_paths_futures ( paths , config ) ; Ok ( stream ) }
Generate futures that load record indexes from files specified by a prefix.
Generate futures that load record indexes from files specified by a prefix .
prefix
P
config
RecordIndexerConfig
prefix
P
config
RecordIndexerConfig