use std::convert;
use o3::{
buffer::{self, hold, pool, pool::owners, segments, storage},
depot::queue::ring,
mem,
};
use sark_core::{
life::session,
manifold::{
server::drain,
stream::{Boundary, Payload, Section},
},
};
use sark_http::field::metadata::{
block, packed,
section::{
rules,
state::{RetainedHead, Selected},
},
};
use crate::{conn::outputs::payloads, flow, frame, frame::numbers, hpack::decode::limit, stream};
#[doc(hidden)]
pub const MIN_RETAINED_WRITE_PREFIX_CAPACITY: usize = frame::HEADER_LEN + payloads::INLINE_PREFIX;
#[must_use]
#[repr(transparent)]
pub struct Ready<'a, T>(ring::Front<'a, ring::Fifo<T>>);
impl<'a, T> Ready<'a, T> {
pub(in crate::conn) const fn new(front: ring::Front<'a, ring::Fifo<T>>) -> Self {
Self(front)
}
}
impl<T> AsRef<T> for Ready<'_, T> {
fn as_ref(&self) -> &T {
self.0.get()
}
}
impl<T> drain::Ready for Ready<'_, T> {
type Item = T;
fn take(self) -> Self::Item {
self.0.remove()
}
}
#[derive(Clone, Copy, Debug)]
pub struct Rejected {
pub admit: limit::Admission,
pub sealed: block::Validation,
pub finished: block::Validation,
}
const _: () = assert!(size_of::<Option<Rejected>>() == 3);
#[derive(Debug)]
pub enum Event<
S = Selected<rules::RetainRaw>,
B = packed::Decoded,
C: session::Storage = storage::Shared,
E: Evidence = Retirement,
> {
PrefaceComplete,
SettingsApplied,
SettingsAck,
Ping {
ack: frame::Ack,
opaque: [u8; 8],
},
GoAway {
last_stream_id: stream::Id,
error: numbers::ErrorCode,
debug: DataPayload<C>,
},
Headers {
stream_id: stream::Key,
head: RetainedHead<B, S, Option<block::Plan>>,
boundary: Boundary,
section: Section,
},
Data {
stream_id: stream::Key,
data: DataPayload<C>,
boundary: Boundary,
},
StreamReset {
stream_id: stream::Key,
error: numbers::ErrorCode,
#[doc(hidden)]
retired: Option<E::Retired>,
validation: Option<Rejected>,
},
EgressRetired {
write: E::Retired,
},
PushPromise {
stream_id: stream::Id,
promised_stream_id: stream::Key,
head: RetainedHead<B, S, Option<block::Plan>>,
},
}
#[must_use = "the returned write owns bytes removed from the connection"]
pub enum Write<E = Retirement, C: session::Storage = storage::Shared> {
Buffered(usize),
Owned {
stream_id: stream::Key,
prefix_len: usize,
body: C,
},
Split {
#[doc(hidden)]
write: E,
prefix_len: usize,
body: C,
},
}
#[doc(hidden)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Retirement {
pub stream_id: stream::Key,
pub charge: hold::Charge,
}
#[doc(hidden)]
pub trait Evidence: Sized {
type Retired: PartialEq + Eq;
#[doc(hidden)]
const OBSERVED: bool;
#[doc(hidden)]
fn project(write: Retirement) -> Option<Self::Retired>;
}
impl Evidence for Retirement {
type Retired = Self;
const OBSERVED: bool = true;
fn project(write: Retirement) -> Option<Self::Retired> {
Some(write)
}
}
#[doc(hidden)]
pub enum Ignore {}
impl Evidence for Ignore {
type Retired = convert::Infallible;
const OBSERVED: bool = false;
fn project(_: Retirement) -> Option<Self::Retired> {
None
}
}
#[must_use = "staged DATA owns flow credit until the machine accepts or cancels it"]
pub(crate) struct Staging {
pub(crate) charge: hold::Charge,
pub(crate) credit: flow::Reserved,
}
impl Staging {
pub(crate) fn into_parts(self) -> (hold::Charge, flow::Reserved) {
(self.charge, self.credit)
}
}
const _: () = assert!(
core::mem::size_of::<Staging>() == core::mem::size_of::<(hold::Charge, flow::Reserved)>()
);
impl Retirement {
pub(crate) fn from_body<C: session::Storage>(stream: stream::Key, body: &C) -> Self {
Self {
stream_id: stream,
charge: hold::Charge::from_owner(
body.as_ref().len(),
buffer::Residency::resident(body),
),
}
}
}
const _: () = {
assert!(core::mem::size_of::<Ignore>() == 0);
assert!(core::mem::size_of::<Option<convert::Infallible>>() == 0);
};
#[derive(Debug)]
pub struct DataPayload<C: session::Storage = storage::Shared> {
segments: segments::Chain<C>,
permit: pool::Span<owners::Plain<pool::Width<1>>, pool::Owned, pool::Frozen>,
}
impl<C: session::Storage> DataPayload<C> {
#[doc(hidden)]
pub fn new(
segments: segments::Chain<C>,
permit: pool::Span<owners::Plain<pool::Width<1>>, pool::Owned, pool::Frozen>,
) -> Self {
Self { segments, permit }
}
pub fn len(&self) -> usize {
self.segments.len()
}
pub fn is_empty(&self) -> bool {
self.segments.is_empty()
}
pub fn count(&self) -> segments::Count {
self.segments.count()
}
pub fn contiguous(&self) -> Option<&[u8]> {
self.segments.contiguous()
}
pub fn segments(&self) -> segments::Iter<'_, C> {
self.segments.segments()
}
pub fn for_each(&self, mut visit: impl FnMut(&[u8])) {
for part in self.segments.segments() {
visit(part.as_ref());
}
}
pub(crate) fn try_append(
mut self,
target: &mut Option<segments::Chain<C>>,
) -> Result<(), mem::Rejected<Self, segments::Error>> {
if self.is_empty() {
self.discard();
return Ok(());
}
match target {
Some(target) => {
if let Err(error) = target.try_append(&mut self.segments) {
return Err(mem::Rejected::new(self, error));
}
self.discard();
}
None => {
let Self { segments, permit } = self;
*target = Some(segments);
drop(permit);
}
}
Ok(())
}
pub fn discard(self) {
drop(self);
}
}
impl<C: session::Storage> Payload<C> for DataPayload<C> {
type Permit = pool::Span<owners::Plain<pool::Width<1>>, pool::Owned, pool::Frozen>;
fn into_parts(self) -> (segments::Chain<C>, Self::Permit) {
(self.segments, self.permit)
}
}