use o3::buffer::resident;
use sark_core::{
body::credit,
decode::{self, exec, request, state},
manifold::schedule::typed,
};
use sark_http::{field::metadata::section::rules, payload::owned};
use crate::{
conn::{self, event, machine},
frame::numbers,
server::{
self, payload, response,
state::exchange::{self, Deferred, Rejected, Routed},
sync::book::{Delayed, Receiving, State, phase::Phase},
},
stream,
};
type Connection<'d, R> = machine::Connection<
crate::ServerRole,
<R as decode::Decode>::Plan,
resident::Shared<'d>,
rules::RejectPush,
event::Ignore,
>;
type Entry<'a, 'd, R> = typed::Entry<
'a,
stream::Key,
Phase<'d, R>,
response::Pending<'d>,
Receiving<'d, R>,
conn::DataBlock,
>;
pub(super) unsafe fn finish<'d, R: decode::Decode>(
entry: Entry<'_, 'd, R>,
account: &mut credit::raw::Account,
conn: &mut Connection<'d, R>,
stream: stream::Key,
limits: server::Config,
) where
Delayed<'d, R>: Deferred<'d, R::Prepared, R::Routes>,
{
match entry {
typed::Entry::Control(entry) => {
let (entry, finished) = entry.change(Phase::try_finish);
if let Some(Err(error)) = finished {
let error = unsafe { close(State::Control(entry.remove()), account, conn) }
.err()
.unwrap_or(error)
.code();
conn.streams().cancel(stream, error);
}
}
typed::Entry::Idle(entry) => {
let result = entry.transition(|input| {
match unsafe { try_body::<R>(input, account) } {
Ok(ready) => {
let blocked = (conn.output().len() >= limits.max_outbound_bytes)
.then_some(conn::DataBlock::Egress);
(
Some(typed::Next::Control(Phase::Request(ready), blocked)),
Ok(()),
)
}
Err(error) => (None, Err(error)),
}
});
if let Err(error) = result {
conn.streams().cancel(stream, error);
}
}
typed::Entry::Data(_) => {}
}
}
pub(super) unsafe fn close<'d, R: decode::Decode>(
value: State<'d, R>,
account: &mut credit::raw::Account,
conn: &mut Connection<'d, R>,
) -> Result<(), payload::Error>
where
Delayed<'d, R>: Deferred<'d, R::Prepared, R::Routes>,
{
if let Some(head) = unsafe { try_close(value, account) }? {
head.discard(conn);
}
Ok(())
}
pub(super) unsafe fn try_push<'d, R: decode::Decode>(
phase: &mut Phase<'d, R>,
account: &mut credit::raw::Account,
data: event::DataPayload<resident::Shared<'d>>,
limits: server::Config,
) -> Result<
payload::Push<event::DataPayload<resident::Shared<'d>>>,
event::DataPayload<resident::Shared<'d>>,
>
where
Delayed<'d, R>: Deferred<'d, R::Prepared, R::Routes>,
{
match phase {
Phase::Routed(routed) => Ok(account.scope(|mut ledger| {
let route =
unsafe { exchange::raw::Route::new(Deferred::route_mut(routed), &mut ledger) };
route.try_push(credit::Disabled::new(), data, limits)
})),
_ => Err(data),
}
}
pub(super) unsafe fn try_close<'d, R: decode::Decode>(
value: State<'d, R>,
account: &mut credit::raw::Account,
) -> Result<Option<response::Head<'d>>, payload::Error>
where
Delayed<'d, R>: Deferred<'d, R::Prepared, R::Routes>,
{
Ok(match value {
State::Control(Phase::Routed(request)) => {
unsafe { discard_route(Deferred::into_route(request), account) }?;
None
}
State::Idle(request) => {
unsafe { discard(Rejected::Body(request.into_body()), account) }?;
None
}
State::Control(Phase::Head(head)) => Some(head),
State::Control(Phase::Request(_) | Phase::Status(_)) | State::Data(_) => None,
})
}
pub(super) unsafe fn discard_route<'d, P, F>(
route: Routed<'d, P, F>,
account: &mut credit::raw::Account,
) -> Result<(), payload::Error>
where
P: request::Prepared,
F: state::Family<P>,
{
account.scope(|mut ledger| {
let route = unsafe { exchange::raw::Routed::new(route, &mut ledger) };
route.discard(credit::Disabled::new())
})
}
pub(super) unsafe fn discard(
input: Rejected<'_>,
account: &mut credit::raw::Account,
) -> Result<(), payload::Error> {
account.scope(|mut ledger| {
let input = unsafe { ledger.bind(input) };
input.discard(&mut ledger, credit::Disabled::new())
})
}
pub(super) unsafe fn try_body<'d, R: decode::Decode>(
input: Receiving<'d, R>,
account: &mut credit::raw::Account,
) -> Result<request::Ready<exec::Sync, R::Prepared, owned::Segments<'d>>, numbers::ErrorCode> {
account.scope(|mut ledger| {
input
.try_map_body(|body| {
let body = unsafe { ledger.bind(body) };
body.try_finish(&mut ledger, credit::Disabled::new())
})
.map(request::Admitted::ready)
.map_err(|failure| {
payload::release_rejected(failure, &mut ledger, &mut credit::Disabled::new())
})
})
}