Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
152 changes: 147 additions & 5 deletions crates/backend/src/opendal.rs
Original file line number Diff line number Diff line change
@@ -1,12 +1,18 @@
/// `OpenDAL` backend for rustic.
use std::{collections::BTreeMap, str::FromStr, sync::OnceLock};
use std::{
collections::BTreeMap,
ffi::OsStr,
str::FromStr,
sync::{Arc, OnceLock},
vec::IntoIter,
};

use bytes::Bytes;
use bytesize::ByteSize;
use log::{error, trace};
use log::{error, trace, warn};
use opendal::{
Metadata,
blocking::Operator,
Entry, Metadata,
blocking::{Operator, StdReader},
layers::{ConcurrentLimitLayer, LoggingLayer, RetryLayer, ThrottleLayer},
options::{ListOptions, ReadOptions},
};
Expand All @@ -15,7 +21,9 @@ use tokio::runtime::Runtime;
use typed_path::UnixPathBuf;

use rustic_core::{
ALL_FILE_TYPES, ErrorKind, FileType, Id, ReadBackend, RusticError, RusticResult, WriteBackend,
ALL_FILE_TYPES, ErrorKind, FileType, Id, ReadBackend, ReadSource, ReadSourceEntry,
ReadSourceOpen, RusticError, RusticResult, WriteBackend,
repofile::{Node, NodeType},
};

mod constants {
Expand Down Expand Up @@ -198,6 +206,43 @@ impl OpenDALBackend {
}
.to_string()
}

/// Turn this `OpenDALBackend into a ReadSource`
///
/// # Errors
/// If listing fails
pub fn as_source(self) -> RusticResult<OpenDALReadSource> {
let list_options = ListOptions {
recursive: true,
..Default::default()
};
// openDAL lister may entries in random order; hence we collect and sort them here.
// This also allows to handle listing errors directly
let mut entries: Vec<_> = self
.operator
.lister_options("", list_options)
.map_err(|err| {
RusticError::with_source(ErrorKind::Backend, "Error listong openDAL source.", err)
})?
.filter_map(|entry| {
// Ignore not needed root path
if let Ok(e) = &entry
&& e.path() == "/"
{
return None;
}
entry
.inspect_err(|err| warn!("ignoring error on openDAL entry: {err}"))
.ok()
})
.collect();
entries.sort_unstable_by(|e1, e2| e1.path().cmp(e2.path()));

Ok(OpenDALReadSource {
entries,
be: Arc::new(self),
})
}
}

impl ReadBackend for OpenDALBackend {
Expand Down Expand Up @@ -572,3 +617,100 @@ mod tests {
Ok(())
}
}

#[derive(Debug)]
/// Describes an open file from the local backend.
pub struct OpenFile(Arc<OpenDALBackend>, String);

impl ReadSourceOpen for OpenFile {
type Reader = StdReader;

/// Open the file from the local backend.
///
/// # Returns
///
/// The read handle to the file from the local backend.
///
/// # Errors
///
/// * If the file could not be opened.
fn open(self) -> RusticResult<Self::Reader> {
let path = self.1;

let reader = || self.0.operator.reader(&path)?.into_std_read(..);

let reader = reader()
.map_err(|err| {
RusticError::with_source(
ErrorKind::InputOutput,
"Failed to open file at `{path}`. Please make sure the file exists and is accessible.",
err,
)
.attach_context("path", path)
})?;
Ok(reader)
}
}

// Walk doesn't implement Debug
#[allow(missing_debug_implementations)]
/// A Lister for a `OpenDALSource`
pub struct OpenDALLister(IntoIter<Entry>, Arc<OpenDALBackend>);

impl Iterator for OpenDALLister {
type Item = RusticResult<ReadSourceEntry<OpenFile>>;

fn next(&mut self) -> Option<Self::Item> {
Ok(self.0.next().map(|e| {
let path = e.path();
// strip "/" suffix from dirs
let path = path.strip_suffix('/').unwrap_or(path);
let name = OsStr::new(e.name());
let metadata = e.metadata();
let node_type = if metadata.is_dir() {
NodeType::Dir
} else {
NodeType::File
};
let meta = rustic_core::repofile::Metadata {
mtime: metadata
.last_modified()
.map(opendal::raw::Timestamp::into_inner),
size: metadata.content_length(),
..Default::default()
};
let node = Node::new_node(name, node_type, meta);
let open = Some(OpenFile(self.1.clone(), path.to_string()));
ReadSourceEntry {
path: path.into(),
node,
open,
}
}))
.transpose()
}
}

#[allow(missing_debug_implementations)]
/// A source to backup using openDAL for the access
pub struct OpenDALReadSource {
entries: Vec<Entry>,
be: Arc<OpenDALBackend>,
}

impl ReadSource for OpenDALReadSource {
type Open = OpenFile;
type Iter = OpenDALLister;
/// Returns the size of the source.
fn size(&self) -> RusticResult<Option<u64>> {
let size = self
.entries
.iter()
.map(|e| e.metadata().content_length())
.sum();
Ok(Some(size))
}
fn entries(&self) -> Self::Iter {
OpenDALLister(self.entries.clone().into_iter(), self.be.clone())
}
}
5 changes: 5 additions & 0 deletions crates/core/src/backend.rs
Original file line number Diff line number Diff line change
Expand Up @@ -431,6 +431,11 @@ impl<O> ReadSourceEntry<O> {
);
Ok(Self { path, node, open })
}

/// Turn this into a tree entry
pub fn as_tree_entry(self) -> (PathBuf, Node) {
(self.path, self.node)
}
}

/// Trait for backends that can read and open sources.
Expand Down
9 changes: 9 additions & 0 deletions crates/core/src/backend/childstdout.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,9 @@ pub struct ChildStdoutSource {

impl ChildStdoutSource {
/// Creates a new `ChildSource`.
///
/// # Errors
/// - if calling the command fails
pub fn new(cmd: &CommandInput, path: PathBuf) -> RusticResult<Self> {
let process = Command::new(cmd.command())
.args(cmd.args())
Expand All @@ -51,6 +54,12 @@ impl ChildStdoutSource {
}

/// Finishes the `ChildSource`
///
/// # Errors
/// - if handling of the return code leads to an error
///
/// # Panics
/// - if the lock for the process cannot be obtained (should not happen)
pub fn finish(self) -> RusticResult<()> {
let status = self.process.lock().unwrap().wait();
self.command
Expand Down
1 change: 1 addition & 0 deletions crates/core/src/backend/stdin.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ pub struct StdinSource {

impl StdinSource {
/// Creates a new `StdinSource`.
#[must_use]
pub const fn new(path: PathBuf) -> Self {
Self { path }
}
Expand Down
Loading
Loading