use std::io::{Result, ErrorKind};
use std::collections::HashSet;
use aws_sdk_s3::Client;
use hyperfile::{BlockPtr, SegmentId};
use hyperfile::staging::{Staging, StagingIntercept, config::StagingConfig, s3::S3Staging};
use hyperfile::meta_loader::s3_batch::S3BlockLoader;
use hyperfile::file::flags::HyperFileFlags;
use hyperfile::file::mode::HyperFileMode;
use hyperfile::config::HyperFileConfig;
use crate::DirStaging;
use crate::file::HyperDirFile;
pub struct HyperDir<'a> {
pub(crate) inner: HyperDirFile<'a, S3Staging, S3BlockLoader>
}
impl<'a> HyperDir<'a> {
pub(crate) async fn do_open_or_create(client: Client, file_config: HyperFileConfig, flags: HyperFileFlags, mode: HyperFileMode, create: bool) -> Result<Self>
{
match Self::open(client.clone(), file_config.clone(), flags.clone()).await {
Ok(hyper) => {
Ok(hyper)
},
Err(e) => {
if create && e.kind() == ErrorKind::NotFound {
return Self::create(client, file_config, flags, mode).await;
}
Err(e)
}
}
}
pub async fn open(client: Client, file_config: HyperFileConfig, flags: HyperFileFlags) -> Result<Self>
{
let dir_staging_config = S3Staging::to_dir_staging_config(&file_config.staging);
let staging = S3Staging::from(&client, dir_staging_config, file_config.runtime.clone()).await?;
let loader = staging.to_block_loader();
let file = HyperDirFile::<S3Staging, S3BlockLoader>::open(staging, loader, file_config, flags).await?;
Ok(Self {
inner: file,
})
}
pub async fn create(client: Client, file_config: HyperFileConfig, flags: HyperFileFlags, mode: HyperFileMode) -> Result<Self>
{
let dir_staging_config = S3Staging::to_dir_staging_config(&file_config.staging);
let staging = S3Staging::create(&client, dir_staging_config, file_config.runtime.clone()).await?;
let loader = staging.to_block_loader();
let file = HyperDirFile::<S3Staging, S3BlockLoader>::new(staging, loader, file_config, flags, mode).await?;
Ok(Self {
inner: file,
})
}
pub async fn create_with_interceptor(client: Client, file_config: HyperFileConfig, flags: HyperFileFlags, mode: HyperFileMode, interceptor: impl StagingIntercept<S3Staging> + 'static) -> Result<Self>
{
let dir_staging_config = S3Staging::to_dir_staging_config(&file_config.staging);
let mut staging = S3Staging::create(&client, dir_staging_config, file_config.runtime.clone()).await?;
staging.interceptor(interceptor);
let loader = staging.to_block_loader();
let file = HyperDirFile::<S3Staging, S3BlockLoader>::new(staging, loader, file_config, flags, mode).await?;
Ok(Self {
inner: file,
})
}
pub async fn stat_fast(client: Client, file_config: HyperFileConfig) -> Result<libc::stat>
{
let dir_staging_config = S3Staging::to_dir_staging_config(&file_config.staging);
let staging = S3Staging::from(&client, dir_staging_config, file_config.runtime.clone()).await?;
HyperDirFile::<S3Staging, S3BlockLoader>::stat_fast(staging).await
}
pub async fn update_stat_fast(client: Client, file_config: HyperFileConfig, stat: &libc::stat) -> Result<libc::stat>
{
let staging = S3Staging::from(&client, file_config.staging.clone(), file_config.runtime.clone()).await?;
HyperDirFile::<S3Staging, S3BlockLoader>::update_stat_fast(staging, stat).await
}
}
impl<'a> HyperDir<'a> {
pub fn staging_config(&self) -> &StagingConfig {
self.inner.staging_config()
}
pub fn with_staging_interceptor(&mut self, i: impl StagingIntercept<S3Staging> + 'static) {
self.inner.staging_interceptor(i)
}
pub async fn fs_reachable_meta_ptrs(&self) -> Result<Vec<BlockPtr>> {
self.inner.reachable_meta_ptrs().await
}
pub async fn fs_reachable_segments(&self) -> Result<HashSet<SegmentId>> {
self.inner.reachable_segments().await
}
}