use std::{fmt::Debug, pin::Pin}; use anyhow::{Error, Result}; use bytes::Bytes; use time::OffsetDateTime; use tokio::io::AsyncWrite; use uuid::Uuid; use crate::store_api::{FileInfo, RawFileInfo}; #[async_trait::async_trait] pub trait DiskAPI: Debug + Send + Sync + 'static { fn is_local(&self) -> bool; fn id(&self) -> Uuid; async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>; async fn read_all(&self, volume: &str, path: &str) -> Result; async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()>; async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>; async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result; async fn append_file(&self, volume: &str, path: &str) -> Result; async fn read_file(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<(Vec, usize)>; async fn rename_data( &self, src_volume: &str, src_path: &str, file_info: &FileInfo, dst_volume: &str, dst_path: &str, ) -> Result; async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>; async fn make_volume(&self, volume: &str) -> Result<()>; async fn stat_volume(&self, volume: &str) -> Result; async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()>; async fn read_version( &self, org_volume: &str, volume: &str, path: &str, version_id: &str, opts: &ReadOptions, ) -> Result; async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result; async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; } pub struct RenameDataResp { pub old_data_dir: String, } #[derive(Debug, Clone, Default)] pub struct DeleteOptions { pub recursive: bool, pub immediate: bool, } #[derive(Debug, Clone)] pub struct ReadMultipleReq { pub bucket: String, pub prefix: String, pub files: Vec, pub max_size: usize, pub metadata_only: bool, pub abort404: bool, pub max_results: usize, } #[derive(Debug, Clone)] pub struct ReadMultipleResp { pub bucket: String, pub prefix: String, pub file: String, pub exists: bool, pub error: String, pub data: Vec, pub mod_time: OffsetDateTime, } impl Default for ReadMultipleResp { fn default() -> Self { Self { bucket: Default::default(), prefix: Default::default(), file: Default::default(), exists: Default::default(), error: Default::default(), data: Default::default(), mod_time: OffsetDateTime::UNIX_EPOCH, } } } pub struct VolumeInfo { pub name: String, pub created: OffsetDateTime, } pub struct ReadOptions { pub read_data: bool, pub healing: bool, } pub struct FileWriter { pub inner: Pin>, } impl AsyncWrite for FileWriter { fn poll_write( mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>, buf: &[u8], ) -> std::task::Poll> { Pin::new(&mut self.inner).poll_write(cx, buf) } fn poll_flush( mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> std::task::Poll> { Pin::new(&mut self.inner).poll_flush(cx) } fn poll_shutdown( mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> std::task::Poll> { Pin::new(&mut self.inner).poll_shutdown(cx) } } impl FileWriter { pub fn new(inner: W) -> Self where W: AsyncWrite + Send + Sync + 'static, { Self { inner: Box::pin(inner) } } } #[derive(Debug, thiserror::Error)] pub enum DiskError { #[error("file not found")] FileNotFound, #[error("file version not found")] FileVersionNotFound, #[error("disk not found")] DiskNotFound, #[error("disk access denied")] FileAccessDenied, #[error("InconsistentDisk")] InconsistentDisk, #[error("volume already exists")] VolumeExists, #[error("unformatted disk error")] UnformattedDisk, #[error("unsupport disk")] UnsupportedDisk, #[error("disk not a dir")] DiskNotDir, #[error("volume not found")] VolumeNotFound, } impl DiskError { pub fn check_disk_fatal_errs(errs: &Vec>) -> Result<()> { println!("errs: {:?}", errs); if Self::count_errs(errs, &DiskError::UnsupportedDisk) == errs.len() { return Err(Error::new(DiskError::UnsupportedDisk)); } // if count_errs(errs, &DiskError::DiskAccessDenied) == errs.len() { // return Err(Error::new(DiskError::DiskAccessDenied)); // } if Self::count_errs(errs, &DiskError::FileAccessDenied) == errs.len() { return Err(Error::new(DiskError::FileAccessDenied)); } // if count_errs(errs, &DiskError::FaultyDisk) == errs.len() { // return Err(Error::new(DiskError::FaultyDisk)); // } if Self::count_errs(errs, &DiskError::DiskNotDir) == errs.len() { return Err(Error::new(DiskError::DiskNotDir)); } // if count_errs(errs, &DiskError::XLBackend) == errs.len() { // return Err(Error::new(DiskError::XLBackend)); // } Ok(()) } pub fn count_errs(errs: &Vec>, err: &DiskError) -> usize { return errs .iter() .filter(|&e| { if e.is_some() { let e = e.as_ref().unwrap(); let cast = e.downcast_ref::(); if cast.is_some() { let cast = cast.unwrap(); return cast == err; } } false }) .count(); } pub fn quorum_unformatted_disks(errs: &Vec>) -> bool { Self::count_errs(errs, &DiskError::UnformattedDisk) >= (errs.len() / 2) + 1 } pub fn is_err(err: &Error, disk_err: &DiskError) -> bool { let cast = err.downcast_ref::(); if cast.is_none() { return false; } let e = cast.unwrap(); e == disk_err } // pub fn match_err(err: Error, matchs: Vec) -> bool { // let cast = err.downcast_ref::(); // if cast.is_none() { // return false; // } // let e = cast.unwrap(); // for i in matchs.iter() { // if e == i { // return true; // } // } // return false; // } } impl PartialEq for DiskError { fn eq(&self, other: &Self) -> bool { core::mem::discriminant(self) == core::mem::discriminant(other) } }