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
38 changes: 19 additions & 19 deletions crates/backend/src/local.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
use std::{
fmt::Debug,
fs::{self, File, Metadata},
io::{Read, Seek, SeekFrom, Write},
io::{Read, Seek, SeekFrom},
path::{Path, PathBuf},
process::Command,
};
Expand All @@ -12,8 +12,8 @@ use log::{debug, error, trace, warn};
use walkdir::WalkDir;

use rustic_core::{
ALL_FILE_TYPES, CommandInput, ErrorKind, FileType, Id, ReadBackend, RusticError, RusticResult,
WriteBackend,
ALL_FILE_TYPES, BytesList, CommandInput, ErrorKind, FileType, Id, ReadBackend, RusticError,
RusticResult, WriteBackend,
};

/// A local backend.
Expand Down Expand Up @@ -470,19 +470,13 @@ impl WriteBackend for LocalBackend {
tpe: FileType,
id: &Id,
_cacheable: bool,
buf: Bytes,
content: BytesList,
) -> RusticResult<()> {
fn write_local_file(filename: &Path, buf: &[u8]) -> RusticResult<()> {
let length = buf.len().try_into().map_err(|err| {
RusticError::with_source(
ErrorKind::Internal,
"Failed to convert length `{length}` to u64.",
err,
)
.attach_context("length", buf.len().to_string())
.ask_report()
})?;

fn write_local_file(
filename: &Path,
mut reader: impl Read,
length: u64,
) -> RusticResult<()> {
let mut file = fs::OpenOptions::new()
.create(true)
.truncate(true)
Expand All @@ -507,7 +501,7 @@ impl WriteBackend for LocalBackend {
.attach_context("path", filename.to_string_lossy())
})?;

file.write_all(buf).map_err(|err| {
_ = std::io::copy(&mut reader, &mut file).map_err(|err| {
RusticError::with_source(
ErrorKind::InputOutput,
"Failed to write to the buffer: `{path}`. Please check the file and try again.",
Expand All @@ -526,7 +520,7 @@ impl WriteBackend for LocalBackend {
Ok(())
}

trace!("writing tpe: {:?}, id: {}", &tpe, &id);
trace!("writing tpe: {tpe:?}, id: {id}");
let filename = self.path(tpe, id);

let parent = self.base_path(tpe, id);
Expand All @@ -545,7 +539,13 @@ impl WriteBackend for LocalBackend {

// Write to temporary file
let filename_tmp = parent.join(Self::filename(tpe, id) + "-tmp-");
match write_local_file(&filename_tmp, &buf) {
let size = content.size();
let reader = content.reader();
match write_local_file(
&filename_tmp,
reader,
size.try_into().expect("size too large for u64"), // should not happen
) {
Ok(file) => file,
Err(err) => {
// Clean-up in case of error
Expand Down Expand Up @@ -585,7 +585,7 @@ impl WriteBackend for LocalBackend {
///
/// * If the file could not be removed.
fn remove(&self, tpe: FileType, id: &Id, _cacheable: bool) -> RusticResult<()> {
trace!("removing tpe: {:?}, id: {}", &tpe, &id);
trace!("removing tpe: {tpe:?}, id: {id}");
let filename = self.path(tpe, id);
fs::remove_file(&filename).map_err(|err|
RusticError::with_source(
Expand Down
10 changes: 5 additions & 5 deletions crates/backend/src/opendal.rs
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ use tokio::runtime::Runtime;
use typed_path::UnixPathBuf;

use rustic_core::{
ALL_FILE_TYPES, ErrorKind, FileType, Id, ReadBackend, ReadSource, ReadSourceEntry,
ALL_FILE_TYPES, BytesList, ErrorKind, FileType, Id, ReadBackend, ReadSource, ReadSourceEntry,
ReadSourceOpen, RusticError, RusticResult, WriteBackend,
repofile::{Node, NodeType},
};
Expand Down Expand Up @@ -498,11 +498,11 @@ impl WriteBackend for OpenDALBackend {
tpe: FileType,
id: &Id,
_cacheable: bool,
buf: Bytes,
content: BytesList,
) -> RusticResult<()> {
trace!("writing tpe: {:?}, id: {}", &tpe, &id);
trace!("writing tpe: {tpe:?}, id: {id}");
let filename = self.path(tpe, id);
_ = self.operator.write(&filename, buf).map_err(|err| {
_ = self.operator.write(&filename, content.into_vec()).map_err(|err| {
RusticError::with_source(
ErrorKind::Backend,
"Writing file `{path}` failed in the backend. Please check if the given path is correct.",
Expand All @@ -524,7 +524,7 @@ impl WriteBackend for OpenDALBackend {
/// * `id` - The id of the file.
/// * `cacheable` - Whether the file is cacheable.
fn remove(&self, tpe: FileType, id: &Id, _cacheable: bool) -> RusticResult<()> {
trace!("removing tpe: {:?}, id: {}", &tpe, &id);
trace!("removing tpe: {tpe:?}, id: {id}");
let filename = self.path(tpe, id);
self.operator.delete(&filename).map_err(|err| {
RusticError::with_source(
Expand Down
10 changes: 8 additions & 2 deletions crates/backend/src/rclone.rs
Original file line number Diff line number Diff line change
Expand Up @@ -376,8 +376,14 @@ impl WriteBackend for RcloneBackend {
/// * `id` - The id of the file.
/// * `cacheable` - Whether the data should be cached.
/// * `buf` - The data to write.
fn write_bytes(&self, tpe: FileType, id: &Id, cacheable: bool, buf: Bytes) -> RusticResult<()> {
self.rest.write_bytes(tpe, id, cacheable, buf)
fn write_bytes(
&self,
tpe: FileType,
id: &Id,
cacheable: bool,
content: rustic_core::BytesList,
) -> RusticResult<()> {
self.rest.write_bytes(tpe, id, cacheable, content)
}

/// Removes the given file.
Expand Down
31 changes: 15 additions & 16 deletions crates/backend/src/rest.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,14 @@ use jiff::SignedDuration;
use log::{trace, warn};
use reqwest::{
Url,
blocking::{Client, ClientBuilder},
blocking::{Body, Client, ClientBuilder},
header::{HeaderMap, HeaderValue},
};
use serde::Deserialize;

use rustic_core::{ErrorKind, FileType, Id, ReadBackend, RusticError, RusticResult, WriteBackend};
use rustic_core::{
BytesList, ErrorKind, FileType, Id, ReadBackend, RusticError, RusticResult, WriteBackend,
};

/// joining URL failed on: `{0}`
#[derive(thiserror::Error, Clone, Copy, Debug, displaydoc::Display)]
Expand Down Expand Up @@ -463,22 +465,19 @@ impl WriteBackend for RestBackend {
tpe: FileType,
id: &Id,
_cacheable: bool,
buf: Bytes,
content: BytesList,
) -> RusticResult<()> {
trace!("writing tpe: {:?}, id: {}", &tpe, &id);
let req_builder = self
.client
.post(
self.url(tpe, id)
.map_err(|err| construct_join_url_error(err, tpe, id, &self.url))?,
)
.body(buf);
trace!("writing tpe: {tpe:?}, id: {id}");
let url = self
.url(tpe, id)
.map_err(|err| construct_join_url_error(err, tpe, id, &self.url))?;

self.retry_notify(|| {
// Note: try_clone() always gives Some(_) as the body is Bytes which is cloneable
_ = req_builder
.try_clone()
.unwrap()
let body = Body::new(content.clone().reader());
_ = self
.client
.post(url.clone())
.body(body)
.send()?
.error_for_status()?;
Ok(())
Expand All @@ -498,7 +497,7 @@ impl WriteBackend for RestBackend {
///
/// * If the backoff failed.
fn remove(&self, tpe: FileType, id: &Id, _cacheable: bool) -> RusticResult<()> {
trace!("removing tpe: {:?}, id: {}", &tpe, &id);
trace!("removing tpe: {tpe:?}, id: {id}");
let url = self
.url(tpe, id)
.map_err(|err| construct_join_url_error(err, tpe, id, &self.url))?;
Expand Down
89 changes: 84 additions & 5 deletions crates/core/src/backend.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ pub(crate) mod warm_up;

use std::{io::Read, ops::Deref, path::PathBuf, sync::Arc};

use bytes::Bytes;
use bytes::{Buf, Bytes, buf::Reader};
use enum_map::Enum;
use log::trace;

Expand Down Expand Up @@ -272,6 +272,73 @@ pub trait FindInBackend: ReadBackend {

impl<T: ReadBackend> FindInBackend for T {}

/// A list of Bytes - used to write to backend
#[derive(Debug, Clone, Default)]
pub struct BytesList(Vec<Bytes>);

impl From<Bytes> for BytesList {
fn from(value: Bytes) -> Self {
Self(vec![value])
}
}
impl From<Vec<u8>> for BytesList {
fn from(value: Vec<u8>) -> Self {
Self(vec![value.into()])
}
}

impl BytesList {
/// Turn this `BytesList` into `Vec<Bytes>`
#[must_use]
pub fn into_vec(self) -> Vec<Bytes> {
self.0
}

/// Get a `Bytes` slice
#[must_use]
pub fn slice(&self) -> &[Bytes] {
&self.0
}

/// Returns the total size of the `BytesList`
pub fn size(&self) -> usize {
self.0.iter().map(Bytes::len).sum()
}

/// Add new `Bytes` to the `BytesList`
pub fn add(&mut self, bytes: Bytes) {
self.0.push(bytes);
}

/// Turn this `BytesList` into a `Read`er
#[must_use]
pub fn reader(self) -> BytesListReader {
let mut remaining = self.0.into_iter();
let reader = remaining.next().unwrap_or_default().reader();
BytesListReader { remaining, reader }
}
}

#[derive(Debug)]
pub struct BytesListReader {
remaining: std::vec::IntoIter<Bytes>,
reader: Reader<Bytes>,
}

impl Read for BytesListReader {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
loop {
match self.reader.read(buf) {
Ok(0) => match self.remaining.next() {
None => break Ok(0),
Some(bytes) => self.reader = bytes.reader(),
},
result => break result,
}
}
}
}

/// Trait for backends that can write.
/// This trait is implemented by all backends that can write data.
pub trait WriteBackend: ReadBackend {
Expand Down Expand Up @@ -304,7 +371,13 @@ pub trait WriteBackend: ReadBackend {
/// # Returns
///
/// The result of the write.
fn write_bytes(&self, tpe: FileType, id: &Id, cacheable: bool, buf: Bytes) -> RusticResult<()>;
fn write_bytes(
&self,
tpe: FileType,
id: &Id,
cacheable: bool,
content: BytesList,
) -> RusticResult<()>;

/// Removes the given file.
///
Expand Down Expand Up @@ -345,7 +418,7 @@ mock! {

impl WriteBackend for Backend {
fn create(&self) -> RusticResult<()>;
fn write_bytes(&self, tpe: FileType, id: &Id, cacheable: bool, buf: Bytes) -> RusticResult<()>;
fn write_bytes(&self, tpe: FileType, id: &Id, cacheable: bool, content: BytesList) -> RusticResult<()>;
fn remove(&self, tpe: FileType, id: &Id, cacheable: bool) -> RusticResult<()>;
}
}
Expand All @@ -354,8 +427,14 @@ impl WriteBackend for Arc<dyn WriteBackend> {
fn create(&self) -> RusticResult<()> {
self.deref().create()
}
fn write_bytes(&self, tpe: FileType, id: &Id, cacheable: bool, buf: Bytes) -> RusticResult<()> {
self.deref().write_bytes(tpe, id, cacheable, buf)
fn write_bytes(
&self,
tpe: FileType,
id: &Id,
cacheable: bool,
content: BytesList,
) -> RusticResult<()> {
self.deref().write_bytes(tpe, id, cacheable, content)
}
fn remove(&self, tpe: FileType, id: &Id, cacheable: bool) -> RusticResult<()> {
self.deref().remove(tpe, id, cacheable)
Expand Down
Loading
Loading