mirrors/rustic
2
0
Fork 0
mirror of https://github.com/rustic-rs/rustic.git synced 2025-10-26 11:18:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

make ReadBackend::list async

Browse Source
This commit is contained in:
Alexander Weiss 2022-03-31 14:26:55 +02:00
parent 710470bfe7
commit bd4edac96b
10 changed files with 61 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/backend/decrypt.rs
View File

@ -28,12 +28,12 @@ pub trait DecryptReadBackend: ReadBackend {
Ok(serde_json::from_slice(&data)?)
}
fn stream_all<F: RepoFile>(
async fn stream_all<F: RepoFile>(
&self,
p: ProgressBar,
// ) -> Result<impl Stream<Item = std::result::Result<F, JoinError>>> {
) -> Result<FuturesUnordered<JoinHandle<(Id, F)>>> {
let list = self.list(F::TYPE)?;
let list = self.list(F::TYPE).await?;
p.set_length(list.len() as u64);
let stream: FuturesUnordered<_> = list
@ -123,12 +123,12 @@ impl<R: ReadBackend, C: CryptoKey> ReadBackend for DecryptBackend<R, C> {
self.backend.location()
}
fn list(&self, tpe: FileType) -> Result<Vec<Id>, Self::Error> {
self.backend.list(tpe)
async fn list(&self, tpe: FileType) -> Result<Vec<Id>, Self::Error> {
self.backend.list(tpe).await
}
fn list_with_size(&self, tpe: FileType) -> Result<Vec<(Id, u32)>, Self::Error> {
self.backend.list_with_size(tpe)
async fn list_with_size(&self, tpe: FileType) -> Result<Vec<(Id, u32)>, Self::Error> {
self.backend.list_with_size(tpe).await
}
async fn read_full(&self, tpe: FileType, id: &Id) -> Result<Vec<u8>, Self::Error> {

4
src/backend/dry_run.rs
View File

@ -45,8 +45,8 @@ impl<BE: DecryptFullBackend> ReadBackend for DryRunBackend<BE> {
self.be.location()
}
fn list_with_size(&self, tpe: FileType) -> Result<Vec<(Id, u32)>, Self::Error> {
self.be.list_with_size(tpe)
async fn list_with_size(&self, tpe: FileType) -> Result<Vec<(Id, u32)>, Self::Error> {
self.be.list_with_size(tpe).await
}
async fn read_full(&self, tpe: FileType, id: &Id) -> Result<Vec<u8>, Self::Error> {

4
src/backend/local.rs
View File

@ -37,7 +37,7 @@ impl ReadBackend for LocalBackend {
self.path.to_str().unwrap()
}
fn list(&self, tpe: FileType) -> Result<Vec<Id>, Self::Error> {
async fn list(&self, tpe: FileType) -> Result<Vec<Id>, Self::Error> {
let walker = WalkDir::new(self.path.join(tpe.name()))
.into_iter()
.filter_map(walkdir::Result::ok)
@ -47,7 +47,7 @@ impl ReadBackend for LocalBackend {
Ok(walker.collect())
}
fn list_with_size(&self, tpe: FileType) -> Result<Vec<(Id, u32)>, Self::Error> {
async fn list_with_size(&self, tpe: FileType) -> Result<Vec<(Id, u32)>, Self::Error> {
let walker = WalkDir::new(self.path.join(tpe.name()))
.into_iter()
.filter_map(walkdir::Result::ok)

59
src/backend/mod.rs
View File

@ -49,11 +49,12 @@ pub trait ReadBackend: Clone + Send + Sync + 'static {
type Error: Send + Sync + std::error::Error + 'static;
fn location(&self) -> &str;
fn list_with_size(&self, tpe: FileType) -> Result<Vec<(Id, u32)>, Self::Error>;
async fn list_with_size(&self, tpe: FileType) -> Result<Vec<(Id, u32)>, Self::Error>;
fn list(&self, tpe: FileType) -> Result<Vec<Id>, Self::Error> {
async fn list(&self, tpe: FileType) -> Result<Vec<Id>, Self::Error> {
Ok(self
.list_with_size(tpe)?
.list_with_size(tpe)
.await?
.into_iter()
.map(|(id, _)| id)
.collect())
@ -68,45 +69,41 @@ pub trait ReadBackend: Clone + Send + Sync + 'static {
length: u32,
) -> Result<Vec<u8>, Self::Error>;
fn find_starts_with(
async fn find_starts_with(
&self,
tpe: FileType,
vec: &[&str],
) -> Result<Vec<Result<Id, anyhow::Error>>, Self::Error> {
self.map_list(tpe, vec, |s, id| id.to_hex().starts_with(s))
.map(|res| {
res.into_iter()
.enumerate()
.map(|(i, id)| match id {
MapResult::Some(id) => Ok(id),
MapResult::None => Err(anyhow!("no suitable id found for {}", &vec[i])),
MapResult::NonUnique => Err(anyhow!("id {} is not unique", &vec[i])),
})
.collect()
})
}
/// map_list
fn map_list<T>(
&self,
tpe: FileType,
vec: &[T],
matches: impl Fn(&T, Id) -> bool,
) -> Result<Vec<MapResult<Id>>, Self::Error> {
let mut res = vec![MapResult::None; vec.len()];
for id in self.list(tpe)? {
let mut results = vec![MapResult::None; vec.len()];
for id in self.list(tpe).await? {
for (i, v) in vec.iter().enumerate() {
if matches(v, id) {
if res[i] == MapResult::None {
res[i] = MapResult::Some(id);
if id.to_hex().starts_with(v) {
if results[i] == MapResult::None {
results[i] = MapResult::Some(id);
} else {
res[i] = MapResult::NonUnique;
results[i] = MapResult::NonUnique;
}
}
}
}
Ok(res)
Ok(results
.into_iter()
.enumerate()
.map(|(i, id)| match id {
MapResult::Some(id) => Ok(id),
MapResult::None => Err(anyhow!("no suitable id found for {}", &vec[i])),
MapResult::NonUnique => Err(anyhow!("id {} is not unique", &vec[i])),
})
.collect())
}
async fn find_id(&self, tpe: FileType, id: &str) -> Result<Id, anyhow::Error> {
Ok(match Id::from_hex(id) {
Ok(id) => id,
// if the given id param is not a full Id, search for a suitable one
Err(_) => self.find_starts_with(tpe, &[&id]).await?.remove(0)?,
})
}
}

5
src/commands/cat.rs
View File

@ -57,10 +57,7 @@ pub(super) async fn execute(be: &impl DecryptReadBackend, opts: Opts) -> Result<
}
async fn cat_file(be: &impl DecryptReadBackend, tpe: FileType, opt: IdOpt) -> Result<()> {
let id = Id::from_hex(&opt.id).or_else(|_| {
// if the given id param is not a full Id, search for a suitable one
be.find_starts_with(tpe, &[&opt.id])?.remove(0)
})?;
let id = be.find_id(tpe, &opt.id).await?;
let data = be.read_encrypted_full(tpe, &id).await?;
println!("{}", String::from_utf8(data)?);

9
src/commands/check.rs
View File

@ -48,7 +48,7 @@ async fn check_packs(be: &impl DecryptReadBackend) -> Result<()> {
let mut packs = HashMap::new();
// TODO: only read index files once
let mut stream = be.stream_all::<IndexFile>(progress_counter())?;
let mut stream = be.stream_all::<IndexFile>(progress_counter()).await?;
while let Some(index) = stream.next().await {
let (_, index_packs) = index?.1.dissolve();
for p in index_packs {
@ -56,7 +56,7 @@ async fn check_packs(be: &impl DecryptReadBackend) -> Result<()> {
}
}
for (id, size) in be.list_with_size(FileType::Pack)? {
for (id, size) in be.list_with_size(FileType::Pack).await? {
match packs.remove(&id) {
None => eprintln!("pack {} not contained in index", id.to_hex()),
Some(index_size) if index_size != size => eprintln!(
@ -82,7 +82,10 @@ async fn check_packs(be: &impl DecryptReadBackend) -> Result<()> {
// check if all snapshots and contained trees can be loaded and contents exist in the index
async fn check_snapshots(index: &(impl IndexedBackend + Unpin)) -> Result<()> {
let mut snap_trees = Vec::new();
let mut stream = index.be().stream_all::<SnapshotFile>(progress_counter())?;
let mut stream = index
.be()
.stream_all::<SnapshotFile>(progress_counter())
.await?;
snap_trees.reserve(stream.size_hint().1.unwrap());
while let Some(snap) = stream.next().await {
snap_trees.push(snap?.1.tree);

4
src/commands/list.rs
View File

@ -17,7 +17,7 @@ pub(super) async fn execute(be: &impl DecryptReadBackend, opts: Opts) -> Result<
let tpe = match opts.tpe.as_str() {
// special treatment for listing blobs: read the index and display it
"blobs" => {
let mut stream = be.stream_all::<IndexFile>(ProgressBar::hidden())?;
let mut stream = be.stream_all::<IndexFile>(ProgressBar::hidden()).await?;
while let Some(index) = stream.next().await {
for pack in index?.1.dissolve().1 {
for blob in pack.blobs() {
@ -34,7 +34,7 @@ pub(super) async fn execute(be: &impl DecryptReadBackend, opts: Opts) -> Result<
t => bail!("invalid type: {}", t),
};
for id in be.list(tpe)? {
for id in be.list(tpe).await? {
println!("{}", id.to_hex());
}

12
src/index/mod.rs
View File

@ -85,8 +85,12 @@ pub struct IndexBackend<BE: DecryptReadBackend> {
impl<BE: DecryptReadBackend> IndexBackend<BE> {
pub async fn new(be: &BE, p: ProgressBar) -> Result<Self> {
v1!("reading index...");
let index =
BoomIndex::full(be.stream_all::<IndexFile>(p.clone())?.map(|i| i.unwrap().1)).await;
let index = BoomIndex::full(
be.stream_all::<IndexFile>(p.clone())
.await?
.map(|i| i.unwrap().1),
)
.await;
p.finish_with_message("done");
Ok(Self {
be: be.clone(),
@ -97,7 +101,9 @@ impl<BE: DecryptReadBackend> IndexBackend<BE> {
pub async fn only_full_trees(be: &BE, p: ProgressBar) -> Result<Self> {
v1!("reading index...");
let index = BoomIndex::only_full_trees(
be.stream_all::<IndexFile>(p.clone())?.map(|i| i.unwrap().1),
be.stream_all::<IndexFile>(p.clone())
.await?
.map(|i| i.unwrap().1),
)
.await;
p.finish_with_message("done");

2
src/repo/keyfile.rs
View File

@ -109,7 +109,7 @@ pub async fn find_key_in_backend<B: ReadBackend>(
match hint {
Some(id) => key_from_backend(be, id, passwd).await,
None => {
for id in be.list(FileType::Key)? {
for id in be.list(FileType::Key).await? {
if let Ok(key) = key_from_backend(be, &id, passwd).await {
return Ok(key);
}

10
src/repo/snapshotfile.rs
View File

@ -84,7 +84,7 @@ impl SnapshotFile {
v1!("getting latest snapshot...");
let mut latest: Option<Self> = None;
let mut pred = predicate;
let mut snaps = be.stream_all::<SnapshotFile>(p.clone())?;
let mut snaps = be.stream_all::<SnapshotFile>(p.clone()).await?;
while let Some((id, mut snap)) = snaps.try_next().await? {
if !pred(&snap) {
@ -105,17 +105,15 @@ impl SnapshotFile {
/// Get a SnapshotFile from the backend by (part of the) id
pub async fn from_id<B: DecryptReadBackend>(be: &B, id: &str) -> Result<Self> {
v1!("getting snapshot...");
let id = Id::from_hex(id).or_else(|_| {
// if the given id param is not a full Id, search for a suitable one
be.find_starts_with(FileType::Snapshot, &[id])?.remove(0)
})?;
let id = be.find_id(FileType::Snapshot, id).await?;
SnapshotFile::from_backend(be, &id).await
}
/// Get all SnapshotFiles from the backend
pub async fn all_from_backend<B: DecryptReadBackend>(be: &B) -> Result<Vec<Self>> {
Ok(be
.stream_all::<SnapshotFile>(ProgressBar::hidden())?
.stream_all::<SnapshotFile>(ProgressBar::hidden())
.await?
.map_ok(|(id, mut snap)| {
snap.set_id(id);
snap