atmos/os_lib/
streams.rs

1//! WHATWG Streams Standard 準拠の Streams API コアモジュール
2//!
3//! `ReadableStream`、`WritableStream`、`TransformStream` の状態機械、
4//! キューイング戦略、バックプレッシャー計算、リーダー/ライターの排他的ロック管理を提供する。
5//! ハードウェア依存のない純粋ロジックであり、ホストテスト環境でも高速・安全に検証可能。
6
7extern crate alloc;
8
9use alloc::collections::VecDeque;
10use alloc::string::String;
11use alloc::vec::Vec;
12
13/// ストリームの動作状態
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub enum StreamState {
16    Readable,
17    Closed,
18    Errored(String),
19}
20
21/// リーダーがストリームから読み出した 1 チャンクの結果
22#[derive(Debug, Clone, PartialEq)]
23pub struct ReadResult<T> {
24    pub value: Option<T>,
25    pub done: bool,
26}
27
28impl<T> ReadResult<T> {
29    pub fn chunk(val: T) -> Self {
30        Self {
31            value: Some(val),
32            done: false,
33        }
34    }
35
36    pub fn done() -> Self {
37        Self {
38            value: None,
39            done: true,
40        }
41    }
42}
43
44/// ストリーム操作のエラー
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub enum StreamError {
47    AlreadyLocked,
48    NotLocked,
49    StreamClosed,
50    StreamErrored(String),
51    CannotEnqueueWhenClosed,
52}
53
54/// ReadableStream の純粋コア実装
55#[derive(Debug, Clone)]
56pub struct ReadableStreamCore<T> {
57    pub state: StreamState,
58    pub queue: VecDeque<T>,
59    pub high_water_mark: f64,
60    pub locked: bool,
61}
62
63impl<T> ReadableStreamCore<T> {
64    pub fn new(high_water_mark: f64) -> Self {
65        Self {
66            state: StreamState::Readable,
67            queue: VecDeque::new(),
68            high_water_mark: if high_water_mark < 0.0 { 1.0 } else { high_water_mark },
69            locked: false,
70        }
71    }
72
73    /// バイト列またはデータ配列から直ちに生成するヘルパー
74    pub fn from_chunks(chunks: Vec<T>) -> Self {
75        let mut stream = Self::new(chunks.len() as f64);
76        for c in chunks {
77            stream.queue.push_back(c);
78        }
79        stream
80    }
81
82    /// バックプレッシャー計算（desiredSize）
83    /// キューが満杯なら 0 以下、受け入れ可能なら正の値を返す
84    pub fn desired_size(&self) -> f64 {
85        match self.state {
86            StreamState::Errored(_) => 0.0,
87            StreamState::Closed => 0.0,
88            StreamState::Readable => self.high_water_mark - (self.queue.len() as f64),
89        }
90    }
91
92    /// チャンクをキューに追加（コントローラの enqueue 処理）
93    pub fn enqueue(&mut self, chunk: T) -> Result<(), StreamError> {
94        match &self.state {
95            StreamState::Readable => {
96                self.queue.push_back(chunk);
97                Ok(())
98            }
99            StreamState::Closed => Err(StreamError::CannotEnqueueWhenClosed),
100            StreamState::Errored(e) => Err(StreamError::StreamErrored(e.clone())),
101        }
102    }
103
104    /// ストリームを正常クローズ（コントローラの close 処理）
105    pub fn close(&mut self) -> Result<(), StreamError> {
106        match &self.state {
107            StreamState::Readable => {
108                self.state = StreamState::Closed;
109                Ok(())
110            }
111            StreamState::Closed => Ok(()),
112            StreamState::Errored(e) => Err(StreamError::StreamErrored(e.clone())),
113        }
114    }
115
116    /// ストリームをエラー状態に遷移
117    pub fn error(&mut self, reason: &str) {
118        self.state = StreamState::Errored(reason.into());
119        self.queue.clear();
120    }
121
122    /// リーダーによる排他ロック取得（getReader）
123    pub fn acquire_reader(&mut self) -> Result<(), StreamError> {
124        if self.locked {
125            Err(StreamError::AlreadyLocked)
126        } else {
127            self.locked = true;
128            Ok(())
129        }
130    }
131
132    /// リーダーのロック解放（releaseLock）
133    pub fn release_reader(&mut self) -> Result<(), StreamError> {
134        if !self.locked {
135            Err(StreamError::NotLocked)
136        } else {
137            self.locked = false;
138            Ok(())
139        }
140    }
141
142    /// リーダーによるチャンク読み出し（read()）
143    pub fn read_chunk(&mut self) -> Result<ReadResult<T>, StreamError> {
144        if let Some(chunk) = self.queue.pop_front() {
145            return Ok(ReadResult::chunk(chunk));
146        }
147
148        match &self.state {
149            StreamState::Readable => {
150                // キューは空だがまだクローズしていない場合（後続の enqueue 待ち）
151                Ok(ReadResult::done())
152            }
153            StreamState::Closed => Ok(ReadResult::done()),
154            StreamState::Errored(e) => Err(StreamError::StreamErrored(e.clone())),
155        }
156    }
157
158    /// ストリームのキャンセル（cancel(reason)）
159    pub fn cancel(&mut self, _reason: &str) {
160        self.queue.clear();
161        self.state = StreamState::Closed;
162    }
163
164    /// キュー内の未読チャンク数
165    pub fn queue_len(&self) -> usize {
166        self.queue.len()
167    }
168
169    /// ReadableStreamCore から WritableStreamCore へデータをパイピングする
170    ///
171    /// 1. 両ストリームの排他的ロックを確保
172    /// 2. キューからチャンクを順次取り出し宛先へ書き込み
173    /// 3. 転送完了後、オプションに応じて宛先のクローズ・両ストリームのロック解除を実施
174    pub fn pipe_to(&mut self, dest: &mut WritableStreamCore<T>, options: PipeOptions) -> Result<(), StreamError> {
175        self.acquire_reader()?;
176        if let Err(e) = dest.acquire_writer() {
177            let _ = self.release_reader();
178            return Err(e);
179        }
180
181        while let Ok(res) = self.read_chunk() {
182            if let Some(chunk) = res.value {
183                if let Err(e) = dest.write(chunk) {
184                    if !options.prevent_cancel {
185                        self.cancel("Pipe destination write error");
186                    }
187                    let _ = self.release_reader();
188                    let _ = dest.release_writer();
189                    return Err(e);
190                }
191            }
192            if res.done || self.queue.is_empty() {
193                break;
194            }
195        }
196
197        if !options.prevent_close && self.state == StreamState::Closed {
198            let _ = dest.close();
199        }
200
201        let _ = self.release_reader();
202        let _ = dest.release_writer();
203
204        Ok(())
205    }
206
207    /// ReadableStreamCore を TransformStreamCore を介して別型の ReadableStreamCore へパイピング
208    pub fn pipe_through<O, F>(&mut self, transform: &mut TransformStreamCore<T, O>, mut transform_fn: F) -> Result<(), StreamError>
209    where
210        F: FnMut(T) -> Vec<O>,
211    {
212        self.acquire_reader()?;
213        while let Ok(res) = self.read_chunk() {
214            if let Some(chunk) = res.value {
215                let out_chunks = transform_fn(chunk);
216                for out in out_chunks {
217                    let _ = transform.readable.enqueue(out);
218                }
219            }
220            if res.done || self.queue.is_empty() {
221                break;
222            }
223        }
224        if self.state == StreamState::Closed {
225            let _ = transform.readable.close();
226        }
227        let _ = self.release_reader();
228        Ok(())
229    }
230}
231
232/// WritableStream の純粋コア実装
233#[derive(Debug, Clone)]
234pub struct WritableStreamCore<T> {
235    pub state: StreamState,
236    pub buffer: Vec<T>,
237    pub locked: bool,
238}
239
240impl<T> WritableStreamCore<T> {
241    pub fn new() -> Self {
242        Self {
243            state: StreamState::Readable,
244            buffer: Vec::new(),
245            locked: false,
246        }
247    }
248
249    pub fn acquire_writer(&mut self) -> Result<(), StreamError> {
250        if self.locked {
251            Err(StreamError::AlreadyLocked)
252        } else {
253            self.locked = true;
254            Ok(())
255        }
256    }
257
258    pub fn release_writer(&mut self) -> Result<(), StreamError> {
259        if !self.locked {
260            Err(StreamError::NotLocked)
261        } else {
262            self.locked = false;
263            Ok(())
264        }
265    }
266
267    pub fn write(&mut self, chunk: T) -> Result<(), StreamError> {
268        match &self.state {
269            StreamState::Readable => {
270                self.buffer.push(chunk);
271                Ok(())
272            }
273            StreamState::Closed => Err(StreamError::StreamClosed),
274            StreamState::Errored(e) => Err(StreamError::StreamErrored(e.clone())),
275        }
276    }
277
278    pub fn close(&mut self) -> Result<(), StreamError> {
279        match &self.state {
280            StreamState::Readable => {
281                self.state = StreamState::Closed;
282                Ok(())
283            }
284            StreamState::Closed => Ok(()),
285            StreamState::Errored(e) => Err(StreamError::StreamErrored(e.clone())),
286        }
287    }
288
289    pub fn abort(&mut self, reason: &str) {
290        self.state = StreamState::Errored(reason.into());
291    }
292}
293
294impl<T> Default for WritableStreamCore<T> {
295    fn default() -> Self {
296        Self::new()
297    }
298}
299
300/// pipeTo の制御オプション
301#[derive(Debug, Clone, PartialEq, Eq)]
302pub struct PipeOptions {
303    /// パイプ完了時に宛先 WritableStream を自動クローズするか（既定: false = クローズする）
304    pub prevent_close: bool,
305    /// エラー発生時に宛先 WritableStream を自動アボートするか（既定: false = アボートする）
306    pub prevent_abort: bool,
307    /// 宛先側でエラー発生時に元 ReadableStream を自動キャンセルするか（既定: false = キャンセルする）
308    pub prevent_cancel: bool,
309}
310
311impl Default for PipeOptions {
312    fn default() -> Self {
313        Self {
314            prevent_close: false,
315            prevent_abort: false,
316            prevent_cancel: false,
317        }
318    }
319}
320
321/// TransformStream の純粋コア実装
322#[derive(Debug, Clone)]
323pub struct TransformStreamCore<I, O> {
324    pub readable: ReadableStreamCore<O>,
325    pub writable: WritableStreamCore<I>,
326}
327
328impl<I, O> TransformStreamCore<I, O> {
329    pub fn new(high_water_mark: f64) -> Self {
330        Self {
331            readable: ReadableStreamCore::new(high_water_mark),
332            writable: WritableStreamCore::new(),
333        }
334    }
335}
atmos/os_lib/streams.rs

atmos/os_lib/
streams.rs
