structured_asic/
payload.rs

1//! Sealed original member input, streamed or already completely CRC-checked.
2
3use crate::{
4    content::Content,
5    zip::{Member, ReadFailure},
6};
7use structured_asn1::{memory::Work, range::RangeSource};
8use structured_cms::stream::{AsyncSource, Receiver};
9
10mod private {
11    pub trait Sealed {}
12}
13impl<S: RangeSource> private::Sealed for Member<'_, '_, S> {}
14impl<S: RangeSource> private::Sealed for Content<'_, '_, S> {}
15
16/// An actual original `ASiC` occurrence. Only the streaming member and its
17/// complete CRC-checked retained content implement this boundary. An arbitrary
18/// slice, caller assertion or independently supplied source cannot implement it.
19pub trait Payload<'v, 's, S: RangeSource>: private::Sealed {
20    /// Exact original acquisition or retained-input stream.
21    type Source<'q>: AsyncSource<Error = ReadFailure<S::Error, S::Document, S::Generation>>
22    where
23        Self: 'q;
24    /// Original complete member identity and producing execution.
25    fn member(&self) -> &Member<'v, 's, S>;
26    /// Preflights the actual source before provider state creation.
27    fn source<W: Work>(&self, work: &W) -> crate::zip::Acquisition<Self::Source<'_>, S>;
28}
29impl<'v, 's, S: RangeSource> Payload<'v, 's, S> for Member<'v, 's, S> {
30    type Source<'q>
31        = crate::zip::Source<'v, 's, S>
32    where
33        Self: 'q;
34    fn member(&self) -> &Member<'v, 's, S> {
35        self
36    }
37    fn source<W: Work>(
38        &self,
39        work: &W,
40    ) -> Result<Self::Source<'_>, ReadFailure<S::Error, S::Document, S::Generation>> {
41        Member::source(self, work)
42    }
43}
44impl<'v, 's, S: RangeSource> Payload<'v, 's, S> for Content<'v, 's, S> {
45    type Source<'q>
46        = Retained<'q, 'v, 's, S>
47    where
48        Self: 'q;
49    fn member(&self) -> &Member<'v, 's, S> {
50        Content::member(self)
51    }
52    fn source<W: Work>(
53        &self,
54        work: &W,
55    ) -> Result<Self::Source<'_>, ReadFailure<S::Error, S::Document, S::Generation>> {
56        let member = self.member();
57        let execution = member.session().query(work);
58        member
59            .archive()
60            .admit(&execution, member.central_span())
61            .map_err(|cause| ReadFailure::Processing {
62                member: member.id(),
63                cause,
64            })?;
65        Ok(Retained { content: self })
66    }
67}
68
69/// Borrowed original immutable decoded allocation. Construction requires an
70/// actual complete member receipt, not an external byte-plus-member pairing.
71pub struct Retained<'q, 'v, 's, S: RangeSource> {
72    content: &'q Content<'v, 's, S>,
73}
74impl<S: RangeSource> AsyncSource for Retained<'_, '_, '_, S> {
75    type Error = ReadFailure<S::Error, S::Document, S::Generation>;
76    #[expect(
77        clippy::unused_async_trait_impl,
78        reason = "defer receiver calls and current admission until polling, preserving the asynchronous stream's cancellation boundary for retained bytes"
79    )]
80    async fn stream<'q, W: Work + 'q, R: Receiver + 'q>(
81        &'q mut self,
82        receiver: &'q mut R,
83        work: &'q W,
84    ) -> Result<(), Self::Error> {
85        let member = self.content.member();
86        let execution = member.session().query(work);
87        let failure = |cause| ReadFailure::Processing {
88            member: member.id(),
89            cause,
90        };
91        member
92            .archive()
93            .admit(&execution, member.central_span())
94            .map_err(failure)?;
95        // Borrowing a chunk does not copy or decode its bytes. The receiver
96        // separately admits its actual hashing/copy work under the same query.
97        for bytes in self.content.bytes().chunks(4096) {
98            execution
99                .charge_work(1)
100                .map_err(|cause| failure(crate::error::resource(member.central_span(), cause)))?;
101            receiver.write(bytes).map_err(ReadFailure::Receiver)?;
102        }
103        member
104            .archive()
105            .admit(&execution, member.central_span())
106            .map_err(failure)?;
107        Ok(())
108    }
109}
structured_asic/payload.rs

structured_asic/
payload.rs
