use crate::{
content::Content,
zip::{Member, ReadFailure},
};
use structured_asn1::{memory::Work, range::RangeSource};
use structured_cms::stream::{AsyncSource, Receiver};
mod private {
pub trait Sealed {}
}
impl<S: RangeSource> private::Sealed for Member<'_, '_, S> {}
impl<S: RangeSource> private::Sealed for Content<'_, '_, S> {}
pub trait Payload<'v, 's, S: RangeSource>: private::Sealed {
type Source<'q>: AsyncSource<Error = ReadFailure<S::Error, S::Document, S::Generation>>
where
Self: 'q;
fn member(&self) -> &Member<'v, 's, S>;
fn source<W: Work>(&self, work: &W) -> crate::zip::Acquisition<Self::Source<'_>, S>;
}
impl<'v, 's, S: RangeSource> Payload<'v, 's, S> for Member<'v, 's, S> {
type Source<'q>
= crate::zip::Source<'v, 's, S>
where
Self: 'q;
fn member(&self) -> &Member<'v, 's, S> {
self
}
fn source<W: Work>(
&self,
work: &W,
) -> Result<Self::Source<'_>, ReadFailure<S::Error, S::Document, S::Generation>> {
Member::source(self, work)
}
}
impl<'v, 's, S: RangeSource> Payload<'v, 's, S> for Content<'v, 's, S> {
type Source<'q>
= Retained<'q, 'v, 's, S>
where
Self: 'q;
fn member(&self) -> &Member<'v, 's, S> {
Content::member(self)
}
fn source<W: Work>(
&self,
work: &W,
) -> Result<Self::Source<'_>, ReadFailure<S::Error, S::Document, S::Generation>> {
let member = self.member();
let execution = member.session().query(work);
member
.archive()
.admit(&execution, member.central_span())
.map_err(|cause| ReadFailure::Processing {
member: member.id(),
cause,
})?;
Ok(Retained { content: self })
}
}
pub struct Retained<'q, 'v, 's, S: RangeSource> {
content: &'q Content<'v, 's, S>,
}
impl<S: RangeSource> AsyncSource for Retained<'_, '_, '_, S> {
type Error = ReadFailure<S::Error, S::Document, S::Generation>;
#[expect(
clippy::unused_async_trait_impl,
reason = "defer receiver calls and current admission until polling, preserving the asynchronous stream's cancellation boundary for retained bytes"
)]
async fn stream<'q, W: Work + 'q, R: Receiver + 'q>(
&'q mut self,
receiver: &'q mut R,
work: &'q W,
) -> Result<(), Self::Error> {
let member = self.content.member();
let execution = member.session().query(work);
let failure = |cause| ReadFailure::Processing {
member: member.id(),
cause,
};
member
.archive()
.admit(&execution, member.central_span())
.map_err(failure)?;
for bytes in self.content.bytes().chunks(4096) {
execution
.charge_work(1)
.map_err(|cause| failure(crate::error::resource(member.central_span(), cause)))?;
receiver.write(bytes).map_err(ReadFailure::Receiver)?;
}
member
.archive()
.admit(&execution, member.central_span())
.map_err(failure)?;
Ok(())
}
}