Compare commits

..

3 Commits

8 changed files with 118 additions and 61 deletions

1
Cargo.lock generated
View File

@ -523,6 +523,7 @@ name = "domani"
version = "0.1.0"
dependencies = [
"acme2",
"async-compression",
"bytes",
"clap",
"env_logger",

View File

@ -49,6 +49,7 @@ hyper-trust-dns = "0.5.0"
gix-hash = "0.14.1"
reqwest = { version = "0.11.23", features = ["gzip", "deflate", "stream"] }
gix-object = "0.41.0"
async-compression = { version = "0.4.6", features = ["tokio", "deflate", "zlib"] }
[patch.crates-io]

View File

@ -52,7 +52,7 @@ impl Error {
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "Unexpected error occurred: {}", self.0)
write!(f, "{}", self.0)
}
}

View File

@ -322,9 +322,9 @@ impl super::Store for FSStore {
// that is cloned.
let data = bytes::Bytes::copy_from_slice(file_object.data.as_slice());
Ok(util::into_box_byte_stream(stream::once(
async move { Ok(data) },
)))
Ok(util::BoxByteStream::from_stream(stream::once(async move {
Ok(data)
})))
}
Kind::Commit | Kind::Tag => Err(unexpected::Error::from(
format!("found object of kind {} in tree", file_object.kind).as_str(),

View File

@ -1,4 +1,4 @@
use crate::error::unexpected::{self, Mappable};
use crate::error::unexpected::{self, Intoable, Mappable};
use crate::{origin, util};
use std::{collections, sync};
@ -7,6 +7,15 @@ struct DescrState {
current_tree: gix_hash::ObjectId,
}
#[derive(thiserror::Error, Clone, Debug, PartialEq)]
enum GetObjectError {
#[error("unavailable due to server-side issue")]
Unavailable,
#[error(transparent)]
Unexpected(#[from] unexpected::Error),
}
#[derive(Default)]
pub struct Proxy {
client: reqwest::Client,
@ -41,12 +50,11 @@ impl Proxy {
.or_unexpected_while("parsing url as reqwest url")?
};
let base_path = match url.to_file_path() {
Ok(path) => path,
Err(()) => return Err(unexpected::Error::from("extracting path from url")),
};
let new_path = base_path.join(sub_path);
let new_path = url
.path()
.parse::<std::path::PathBuf>()
.or_unexpected_while("parsing url path")?
.join(sub_path);
url.set_path(
new_path
@ -64,7 +72,7 @@ impl Proxy {
let (url, branch_name) = Self::deconstruct_descr(descr);
let refs_url =
Self::construct_url(url, "/info/refs").or_unexpected_while("constructing refs url")?;
Self::construct_url(url, "info/refs").or_unexpected_while("constructing refs url")?;
// when fetching refs we assume that any issue indicates that the origin itself
// (and therefore the URL) has some kind of issue.
@ -102,24 +110,49 @@ impl Proxy {
&self,
descr: &origin::Descr,
oid: &gix_hash::ObjectId,
) -> unexpected::Result<Option<reqwest::Response>> {
expect_kind: gix_object::Kind,
) -> Result<util::BoxByteStream, GetObjectError> {
let hex = oid.to_string();
let (url, _) = Self::deconstruct_descr(descr);
let object_url = Self::construct_url(
url,
format!("/objects/{}/{}", &hex[..2], &hex[2..]).as_str(),
)
let object_url =
Self::construct_url(url, format!("objects/{}/{}", &hex[..2], &hex[2..]).as_str())
.or_unexpected_while("constructing refs url")?;
Ok(self
let mut loose_object = self
.client
.get(object_url)
.send()
.await
.or_unexpected_while("performing request")?
.or(Err(GetObjectError::Unavailable))?
.error_for_status()
.ok())
.map(|res| {
use async_compression::tokio::bufread::ZlibDecoder;
use futures::stream::TryStreamExt;
use std::io;
tokio::io::BufReader::new(ZlibDecoder::new(tokio_util::io::StreamReader::new(
res.bytes_stream()
.map_err(|e| io::Error::new(io::ErrorKind::Other, e)),
)))
})
.or(Err(GetObjectError::Unavailable))?;
use tokio::io::AsyncBufReadExt;
let mut header = Vec::<u8>::new();
loose_object
.read_until(0, &mut header)
.await
.or(Err(GetObjectError::Unavailable))?;
let (kind, _, _) =
gix_object::decode::loose_header(&header).or(Err(GetObjectError::Unavailable))?;
if kind != expect_kind {
return Err(GetObjectError::Unavailable);
}
Ok(util::BoxByteStream::from_async_read(loose_object))
}
async fn get_commit_tree(
@ -128,17 +161,18 @@ impl Proxy {
commit_hash: &gix_hash::ObjectId,
) -> Result<gix_hash::ObjectId, origin::SyncError> {
let commit_object_bytes = self
.get_object(descr, commit_hash)
.await?
.ok_or(origin::SyncError::Unavailable)?
.bytes()
.get_object(descr, commit_hash, gix_object::Kind::Commit)
.await
.map_err(|e| match e {
GetObjectError::Unavailable => origin::SyncError::Unavailable,
GetObjectError::Unexpected(_) => e.into_unexpected().into(),
})?
.read_to_end()
.await
.or(Err(origin::SyncError::Unavailable))?;
let commit_object = gix_object::ObjectRef::from_loose(commit_object_bytes.as_ref())
.or(Err(origin::SyncError::Unavailable))?
.into_commit()
.ok_or(origin::SyncError::Unavailable)?;
let commit_object = gix_object::CommitRef::from_bytes(commit_object_bytes.as_ref())
.or(Err(origin::SyncError::Unavailable))?;
Ok(commit_object.tree())
}
@ -150,17 +184,18 @@ impl Proxy {
entry_name: &str,
) -> Result<gix_object::tree::Entry, origin::GetFileError> {
let tree_object_bytes = self
.get_object(descr, tree_hash)
.await?
.ok_or(origin::GetFileError::Unavailable)?
.bytes()
.get_object(descr, tree_hash, gix_object::Kind::Tree)
.await
.map_err(|e| match e {
GetObjectError::Unavailable => origin::GetFileError::Unavailable,
GetObjectError::Unexpected(_) => e.into_unexpected().into(),
})?
.read_to_end()
.await
.or(Err(origin::GetFileError::Unavailable))?;
let tree_object = gix_object::ObjectRef::from_loose(tree_object_bytes.as_ref())
.or(Err(origin::GetFileError::Unavailable))?
.into_tree()
.ok_or(origin::GetFileError::Unavailable)?;
let tree_object = gix_object::TreeRef::from_bytes(tree_object_bytes.as_ref())
.or(Err(origin::GetFileError::Unavailable))?;
for entry in tree_object.entries {
if entry.filename == entry_name {
@ -227,9 +262,7 @@ impl origin::Store for Proxy {
let path = path
.as_str()
.parse::<std::path::PathBuf>()
.or_unexpected_while("parsing path")?
.canonicalize()
.or_unexpected_while("canonicalizing path")?;
.or_unexpected_while("parsing path")?;
let path_parts = path.iter().collect::<Vec<&std::ffi::OsStr>>();
@ -288,19 +321,14 @@ impl origin::Store for Proxy {
.into());
}
let res = self
.get_object(&descr, &entry.oid)
.await?
.map_unexpected_while(|| format!("object for entry {:?} not found", entry))?
.bytes_stream();
use futures::StreamExt;
Ok(util::into_box_byte_stream(res.map(|r| {
use std::io::{Error, ErrorKind};
r.map_err(|e| Error::new(ErrorKind::ConnectionAborted, e))
})))
// TODO this is still not correct, as it will include the git object header
self.get_object(&descr, &entry.oid, gix_object::Kind::Blob)
.await
.map_err(|e| match e {
GetObjectError::Unavailable => origin::GetFileError::Unavailable,
GetObjectError::Unexpected(_) => e
.into_unexpected_while(format!("getting object for entry {:?}", entry))
.into(),
})
})
}
}

View File

@ -62,7 +62,7 @@ impl Service {
w.write_all("\r\n".as_bytes()).await.or_unexpected()?;
if let Some(body) = body {
let mut body = tokio_util::io::StreamReader::new(body);
let mut body = body.into_async_read();
copy(&mut body, &mut w).await.or_unexpected()?;
}

View File

@ -219,7 +219,7 @@ impl Service {
use domain::manager::GetFileError;
match self.domain_manager.get_file(&settings, &path).await {
Ok(f) => self.serve(200, &path, Body::wrap_stream(f)),
Ok(f) => self.serve(200, &path, Body::wrap_stream(f.into_stream())),
Err(GetFileError::FileNotFound) => self.render_error_page(404, "File not found"),
Err(GetFileError::Unavailable) => self.render_error_page(502, "Content unavailable"),
Err(GetFileError::DescrNotSynced) => self.internal_error(

View File

@ -1,3 +1,4 @@
use futures::stream::BoxStream;
use std::{fs, io, path, pin};
pub fn open_file(path: &path::Path) -> io::Result<Option<fs::File>> {
@ -31,13 +32,39 @@ pub fn parse_file<T: std::str::FromStr>(
}
}
pub type BoxByteStream = futures::stream::BoxStream<'static, io::Result<bytes::Bytes>>;
pub struct BoxByteStream(BoxStream<'static, io::Result<bytes::Bytes>>);
pub fn into_box_byte_stream<T>(v: T) -> BoxByteStream
where
T: futures::stream::Stream<Item = std::io::Result<bytes::Bytes>> + Send + 'static,
{
Box::into_pin(Box::new(v))
impl BoxByteStream {
pub fn from_stream<S>(s: S) -> Self
where
S: futures::stream::Stream<Item = std::io::Result<bytes::Bytes>> + Send + 'static,
{
Self(Box::into_pin(Box::new(s)))
}
pub fn from_async_read<R>(r: R) -> Self
where
R: tokio::io::AsyncRead + Send + 'static,
{
Self::from_stream(tokio_util::io::ReaderStream::new(r))
}
pub fn into_stream(
self,
) -> impl futures::stream::Stream<Item = std::io::Result<bytes::Bytes>> + Send + 'static {
self.0
}
pub fn into_async_read(self) -> impl tokio::io::AsyncRead + Send + 'static {
tokio_util::io::StreamReader::new(self.into_stream())
}
pub async fn read_to_end(self) -> io::Result<Vec<u8>> {
use tokio::io::AsyncReadExt;
let mut buf = Vec::<u8>::new();
self.into_async_read().read_to_end(&mut buf).await?;
Ok(buf)
}
}
pub type BoxFuture<'a, O> = pin::Pin<Box<dyn futures::Future<Output = O> + Send + 'a>>;