structured_cms/
stream.rs

1//! One payload pass with sticky source, callback and resource refusal.
2
3use crate::Error;
4use structured_asn1::memory::Work;
5
6/// A write refusal whose original cause remains held by the stream driver.
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub struct Refused;
9
10/// A source-facing chunk receiver.
11pub trait Receiver {
12    /// Delivers the next complete chunk in logical order.
13    ///
14    /// On refusal the source must stop. The driver additionally enforces sticky
15    /// refusal if a source ignores this marker, without cloning the cause.
16    fn write(&mut self, bytes: &[u8]) -> core::result::Result<(), Refused>;
17}
18
19/// Caller-supplied payload acquisition; successful return denotes complete EOF.
20///
21/// Acquisition must admit its own data-dependent work/storage before execution
22/// through the supplied domain. This trait neither fetches document references
23/// nor proves accounting inside arbitrary caller code.
24pub trait Source {
25    /// The original acquisition failure, distinct from a signature mismatch.
26    type Error;
27
28    /// Feeds every chunk once, returning an error for an incomplete source.
29    fn stream<W: Work>(
30        &mut self,
31        receiver: &mut impl Receiver,
32        execution: &W,
33    ) -> core::result::Result<(), Self::Error>;
34}
35
36/// Feeds original ranges, including the SET tag and original signedAttrs
37/// remainder returned by [`crate::attribute::Attributes::signed_bytes`].
38/// No concatenation or canonical re-encoding occurs; each metadata step is
39/// admitted in the current operation before dispatch.
40impl<const N: usize> Source for [&[u8]; N] {
41    type Error = RangesError;
42
43    fn stream<W: Work>(
44        &mut self,
45        receiver: &mut impl Receiver,
46        execution: &W,
47    ) -> Result<(), Self::Error> {
48        for bytes in self.iter() {
49            execution.charge_work(1).map_err(RangesError::Resource)?;
50            receiver.write(bytes).map_err(RangesError::Receiver)?;
51        }
52        Ok(())
53    }
54}
55
56/// Range acquisition has no ASN.1 source location of its own.
57#[derive(Debug)]
58pub enum RangesError {
59    /// Current work admission failed before walking a range.
60    Resource(structured_asn1::memory::MemoryError),
61    /// The driver holds the original sink refusal.
62    Receiver(Refused),
63}
64
65/// An injected processing sink, such as a set of digest/raw verification states.
66pub trait Sink {
67    /// The callback's original failure, preserving provider/resource provenance.
68    type Error;
69
70    /// Processes a chunk after admitting its complete transitive operation.
71    ///
72    /// The driver pays dispatch; the callback pays actual hashing, copying,
73    /// opaque state growth and fan-out before execution. No Clone/Send/Sync
74    /// bound or particular cryptographic engine is required.
75    fn write<W: Work>(
76        &mut self,
77        bytes: &[u8],
78        execution: &W,
79    ) -> core::result::Result<(), Self::Error>;
80}
81
82/// Why a payload pass could not finish.
83#[derive(Debug)]
84pub enum StreamError<SE, CE> {
85    /// The source did not reach successful EOF.
86    Source(SE),
87    /// A sink callback refused, with its original cause.
88    Callback(CE),
89    /// Common processing refused resources before dispatch.
90    Processing(Error),
91}
92
93enum Failure<E> {
94    Callback(E),
95    Processing(Error),
96}
97
98struct Guard<'a, S: Sink, W: Work> {
99    sink: &'a mut S,
100    execution: &'a W,
101    field: &'static str,
102    at: u64,
103    failure: Option<Failure<S::Error>>,
104}
105
106impl<S: Sink, W: Work> Receiver for Guard<'_, S, W> {
107    fn write(&mut self, bytes: &[u8]) -> core::result::Result<(), Refused> {
108        if self.failure.is_some() {
109            return Err(Refused);
110        }
111        if let Err(reason) = self.execution.charge_work(1) {
112            self.failure = Some(Failure::Processing(Error::resource(
113                reason, self.field, self.at,
114            )));
115            return Err(Refused);
116        }
117        if let Err(error) = self.sink.write(bytes, self.execution) {
118            self.failure = Some(Failure::Callback(error));
119            return Err(Refused);
120        }
121        Ok(())
122    }
123}
124
125/// Feeds one source once and preserves the first sink refusal without cloning.
126///
127/// `field`/`at` locate processing in the caller's enclosing structure; they do
128/// not invent an ASN.1 position for detached bytes. A callback refusal takes
129/// precedence over a later source failure caused by that refusal. A partial
130/// source never returns success, even if a prefix reached the sink.
131pub fn run<S: Source, C: Sink, W: Work>(
132    source: &mut S,
133    sink: &mut C,
134    execution: &W,
135    field: &'static str,
136    at: u64,
137) -> core::result::Result<(), StreamError<S::Error, C::Error>> {
138    execution
139        .charge_work(1)
140        .map_err(|reason| StreamError::Processing(Error::resource(reason, field, at)))?;
141    let mut guard = Guard {
142        sink,
143        execution,
144        field,
145        at,
146        failure: None,
147    };
148    let acquired = source.stream(&mut guard, execution);
149    match guard.failure {
150        Some(Failure::Callback(error)) => Err(StreamError::Callback(error)),
151        Some(Failure::Processing(error)) => Err(StreamError::Processing(error)),
152        None => acquired.map_err(StreamError::Source),
153    }
154}
structured_cms/stream.rs

structured_cms/
stream.rs
