You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
370 lines
9.2 KiB
370 lines
9.2 KiB
//! Handles suspending and reloading posts
|
|
use super::*;
|
|
use std::{
|
|
marker::{
|
|
PhantomData,
|
|
Unpin,
|
|
Send,
|
|
Sync,
|
|
},
|
|
io,
|
|
collections::HashMap,
|
|
ops::Range,
|
|
borrow::{
|
|
Cow,
|
|
Borrow,
|
|
},
|
|
error,
|
|
fmt,
|
|
};
|
|
use tokio::{
|
|
prelude::*,
|
|
io::{
|
|
AsyncRead,
|
|
AsyncWrite,
|
|
},
|
|
};
|
|
use bytes::{
|
|
IntoBytes,
|
|
FromBytes,
|
|
};
|
|
|
|
/// Represents an opaque bytes stream of serialised data to insert into `SuspendStream`.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct Object
|
|
{
|
|
data: Vec<u8>,
|
|
data_instances: HashMap<Cow<'static, str>, Range<usize>>,
|
|
}
|
|
|
|
fn calc_len<T,I,U>(from: I) -> U
|
|
where I: IntoIterator<Item = T>,
|
|
T: Borrow<Range<U>>,
|
|
U: std::cmp::Ord + Default + Copy,
|
|
{
|
|
let mut max = Default::default();
|
|
|
|
for i in from.into_iter()
|
|
{
|
|
let i = i.borrow().end;
|
|
if i > max {
|
|
max = i
|
|
}
|
|
}
|
|
max
|
|
}
|
|
|
|
impl Object
|
|
{
|
|
/// Create a new empty `Object`.
|
|
#[inline] pub fn new() -> Self
|
|
{
|
|
Self{data: Vec::new(), data_instances: HashMap::new()}
|
|
}
|
|
|
|
/// Are the internal mappings of this object valid?
|
|
pub fn validate(&self) -> bool
|
|
{
|
|
let len = self.data.len();
|
|
for (_, range) in self.data_instances.iter() {
|
|
if range.start + range.end > len {
|
|
return false;
|
|
}
|
|
}
|
|
true
|
|
}
|
|
|
|
/// Insert a boolean
|
|
pub fn insert_bool(&mut self, name: impl Into<Cow<'static, str>>, value: bool)
|
|
{
|
|
self.insert_bytes(name, if value {&[1]} else {&[0]})
|
|
}
|
|
/// Get a boolean
|
|
pub fn get_bool(&self, name: impl Borrow<str>) -> Option<bool>
|
|
{
|
|
self.get_bytes(name).map(|x| if x[0]==0 {false} else {true})
|
|
}
|
|
|
|
/// Insert a UTF-8 string
|
|
pub fn insert_string(&mut self, name: impl Into<Cow<'static, str>>, stri: impl AsRef<str>)
|
|
{
|
|
self.insert_bytes(name, stri.as_ref().as_bytes())
|
|
}
|
|
|
|
/// Try to get a UTF-8 string, returning `None` if either the name was not present, or the string was not formatted correctly
|
|
pub fn try_get_string(&self, name: impl Borrow<str>) -> Option<&str>
|
|
{
|
|
if let Some(bytes) = self.get_bytes(name) {
|
|
std::str::from_utf8(bytes).ok()
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
/// Try to get a UTF-8 string, returning `None` if the name was not present
|
|
///
|
|
/// # Panics
|
|
/// If the present string was not valid UTF-8
|
|
pub fn get_string(&self, name: impl Borrow<str>) -> Option<&str>
|
|
{
|
|
if let Some(bytes) = self.get_bytes(name) {
|
|
Some(std::str::from_utf8(bytes).expect("String contained invalid UTF-8"))
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
/// Insert bytes directly with this name
|
|
pub fn insert_bytes(&mut self, name: impl Into<Cow<'static, str>>, bytes: impl AsRef<[u8]>)
|
|
{
|
|
let bytes= bytes.as_ref();
|
|
let start = self.data.len();
|
|
self.data.extend_from_slice(bytes);
|
|
let end = self.data.len();
|
|
|
|
self.data_instances.insert(name.into(), start..end).unwrap_none();
|
|
}
|
|
|
|
/// Insert a value's bytes directly with this name
|
|
pub unsafe fn insert_value_raw<T,U>(&mut self, name: U, value: &T)
|
|
where T: ?Sized,
|
|
U: Into<Cow<'static, str>>
|
|
{
|
|
self.insert_bytes(name, bytes::refer(value))
|
|
}
|
|
|
|
/// Convert a value to bytes and insert it into this object
|
|
pub fn insert_value<T: IntoBytes, U>(&mut self, name: U, value: T)
|
|
where U: Into<Cow<'static, str>>
|
|
{
|
|
self.insert_bytes(name, value.into_bytes())
|
|
}
|
|
|
|
|
|
/// Try to get a value from the bytes specified by name
|
|
pub fn get_value<T>(&self, name: impl Borrow<str>) -> Result<Option<T>, T::Error>
|
|
where T: FromBytes
|
|
{
|
|
if let Some(range) = self.data_instances.get(name.borrow()) {
|
|
Ok(Some(T::from_bytes(&self.data[range.clone()])?))
|
|
} else {
|
|
Ok(None)
|
|
}
|
|
}
|
|
|
|
/// Try to get the bytes specified by name
|
|
pub fn get_bytes(&self, name: impl Borrow<str>) -> Option<&[u8]>
|
|
{
|
|
if let Some(range) = self.data_instances.get(name.borrow()) {
|
|
Some(&self.data[range.clone()])
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
/// Try to get the value spcified by name
|
|
///
|
|
/// # Panics
|
|
/// If `T` cannot fit into the size of the range
|
|
pub unsafe fn get_value_raw<T>(&self, name: impl Borrow<str>) -> Option<&T>
|
|
{
|
|
self.get_bytes(name).map(|x| bytes::derefer(x))
|
|
}
|
|
|
|
/// Try to get the value specified by name. Will return `None` if `T` cannot fit in the returned bytes.
|
|
pub unsafe fn try_get_value_raw<T>(&self, name: impl Borrow<str>) -> Option<&T>
|
|
{
|
|
if let Some(bytes) = self.get_bytes(name) {
|
|
if bytes.len() >= std::mem::size_of::<T>() {
|
|
Some(bytes::derefer(bytes))
|
|
} else {
|
|
#[cfg(debug_assertions)] eprintln!("Warning! Likely data corruption as {} (size {}) cannot fit into {} bytes",std::any::type_name::<T>(), std::mem::size_of::<T>(), bytes.len());
|
|
None
|
|
}
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
/// Consume into the data
|
|
#[deprecated = "Invalid ABI"] fn into_bytes(self) -> Box<[u8]>
|
|
{
|
|
let mut output = Vec::new(); //TOOO: Get cap
|
|
|
|
debug_assert!(self.validate(), "passing invalid object to serialise");
|
|
|
|
unsafe {
|
|
output.extend_from_slice(bytes::refer(&self.data_instances.len()));
|
|
for (name, Range{start, end}) in self.data_instances.into_iter() {
|
|
let name = name.as_bytes();
|
|
|
|
output.extend_from_slice(bytes::refer(&name.len()));
|
|
output.extend_from_slice(name);
|
|
|
|
output.extend_from_slice(bytes::refer(&start));
|
|
output.extend_from_slice(bytes::refer(&end));
|
|
}
|
|
}
|
|
|
|
output.extend(self.data);
|
|
output.into_boxed_slice()
|
|
}
|
|
|
|
/// Read an object from a stream
|
|
pub async fn from_stream<T>(input: &mut T) -> io::Result<Self>
|
|
where T: AsyncRead + Unpin + ?Sized
|
|
{
|
|
let mut ext_buf = Vec::new();
|
|
|
|
macro_rules! bin {
|
|
(usize) => {
|
|
{
|
|
use std::convert::TryFrom;
|
|
let value: u64 = input.read_u64().await?;
|
|
usize::try_from(value).map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "u64 cannot fit in usize"))?
|
|
}
|
|
};
|
|
($size:expr) => {
|
|
{
|
|
let sz = $size;
|
|
ext_buf.resize(sz, 0);
|
|
assert_eq!(input.read_exact(&mut ext_buf[..sz]).await?, ext_buf.len());
|
|
&ext_buf[..sz]
|
|
}
|
|
}
|
|
}
|
|
let entries = bin!(usize);
|
|
|
|
let mut instances = HashMap::with_capacity(entries);
|
|
|
|
for _i in 0..entries
|
|
{
|
|
let name_len = bin!(usize);
|
|
|
|
let name_bytes = bin!(name_len);
|
|
let name_str = std::str::from_utf8(name_bytes).map_err(|_| io::Error::new(io::ErrorKind::InvalidData,"item name was corrupted"))?;
|
|
|
|
let start = bin!(usize);
|
|
let end = bin!(usize);
|
|
|
|
instances.insert(Cow::Owned(name_str.to_owned()), start..end);
|
|
}
|
|
|
|
let expected_len = bin!(usize);
|
|
if expected_len != calc_len(instances.iter().map(|(_, v)| v))
|
|
{
|
|
return Err(io::Error::new(io::ErrorKind::InvalidData, "expected and read sizes differ"));
|
|
}
|
|
|
|
let mut data = vec![0; expected_len];
|
|
assert_eq!(input.read_exact(&mut data[..]).await?, expected_len);
|
|
|
|
Ok(Self {
|
|
data,
|
|
data_instances: instances,
|
|
})
|
|
}
|
|
|
|
/// Write this instance into an async stream
|
|
pub async fn into_stream<T>(&self, output: &mut T) -> io::Result<usize>
|
|
where T: AsyncWrite + Unpin + ?Sized
|
|
{
|
|
debug_assert!(self.validate(), "passing invalid object to serialise");
|
|
|
|
let mut written=0usize;
|
|
macro_rules! bin {
|
|
($bytes:expr) => {
|
|
{
|
|
let bytes = $bytes;
|
|
output.write_all(bytes).await?;
|
|
written+=bytes.len();
|
|
}
|
|
};
|
|
(usize $value:expr) => {
|
|
{
|
|
use std::convert::TryInto;
|
|
let val: u64 = $value.try_into().map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "size cannot fit in u64"))?;
|
|
output.write_u64(val).await?;
|
|
written += std::mem::size_of::<u64>();
|
|
}
|
|
};
|
|
}
|
|
|
|
unsafe {
|
|
bin!(usize self.data_instances.len());
|
|
for (name, &Range{start, end}) in self.data_instances.iter() {
|
|
let name = name.as_bytes();
|
|
|
|
bin!(usize name.len());
|
|
bin!(name);
|
|
|
|
bin!(usize start);
|
|
bin!(usize end);
|
|
}
|
|
}
|
|
|
|
bin!(usize self.data.len()); //for additional checks
|
|
bin!(&self.data);
|
|
Ok(written)
|
|
}
|
|
}
|
|
|
|
/// A suspend stream represents a stream of objects of _the same type_. Can be any number of them, but they all must be for the same type.
|
|
#[async_trait]
|
|
pub trait SuspendStream
|
|
{
|
|
/// Write an object into the opaque stream.
|
|
async fn set_object(&mut self, obj: Object) -> Result<(), Error>;
|
|
/// Read an object from the opaque stream.
|
|
async fn get_object(&mut self) -> Result<Option<Object>, Error>;
|
|
}
|
|
|
|
/// An error that occoured in a suspend operation
|
|
#[derive(Debug)]
|
|
#[non_exhaustive]
|
|
pub enum Error {
|
|
BadObject,
|
|
Corruption,
|
|
IO(io::Error),
|
|
|
|
Unknown,
|
|
}
|
|
impl error::Error for Error
|
|
{
|
|
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
|
|
Some(match &self {
|
|
Self::IO(io) => io,
|
|
_ => return None,
|
|
})
|
|
}
|
|
}
|
|
impl fmt::Display for Error
|
|
{
|
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result
|
|
{
|
|
match self {
|
|
Self::BadObject => write!(f, "unexpected object in stream"),
|
|
Self::Corruption => write!(f, "data stream corruption"),
|
|
Self::IO(_) => write!(f, "i/o error"),
|
|
_ => write!(f, "unknown error"),
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
/// A suspendable type, that can save and reload its data atomically
|
|
#[async_trait]
|
|
pub trait Suspendable: Sized
|
|
{
|
|
async fn suspend<S: SuspendStream + Send + Sync+ ?Sized>(self, into: &mut S) -> Result<(), Error>;
|
|
async fn load<S: SuspendStream + Send+ Sync+?Sized>(from: &mut S) -> Result<Self, Error>;
|
|
}
|
|
|
|
|
|
/*
|
|
pub struct SuspenceState<T>
|
|
where T: Suspendable
|
|
{
|
|
|
|
_phantom: PhantomData<T>,
|
|
}*/
|