ferrijs_std/stream_web/readable/stream/
algorithms.rs

1use std::{cell::RefCell, rc::Rc};
2
3use crate::utils::option::{Null, Undefined};
4use crate::utils::primordials::Primordial;
5use rquickjs::{
6    class::Trace, prelude::This, Class, Ctx, Function, JsLifetime, Object, Promise, Result, Value,
7};
8
9use crate::stream_web::{
10    readable::controller::ReadableStreamControllerClass,
11    transform::{
12        controller::TransformStreamDefaultControllerClass,
13        stream::{self as transform_stream, TransformStreamClass},
14    },
15    utils::promise::{promise_resolved_with, PromisePrimordials},
16};
17
18use super::tee::TeeState;
19
20#[derive(Clone)]
21pub enum StartAlgorithm<'js> {
22    ReturnUndefined,
23    Function {
24        f: Function<'js>,
25        underlying_source: Null<Undefined<Object<'js>>>,
26    },
27}
28
29impl<'js> StartAlgorithm<'js> {
30    pub(crate) fn call(
31        &self,
32        ctx: Ctx<'js>,
33        controller: ReadableStreamControllerClass<'js>,
34    ) -> Result<Value<'js>> {
35        match self {
36            StartAlgorithm::ReturnUndefined => Ok(Value::new_undefined(ctx.clone())),
37            StartAlgorithm::Function {
38                f,
39                underlying_source,
40            } => f.call::<_, Value>((This(underlying_source.clone()), controller)),
41        }
42    }
43}
44
45type PullRustFn<'js> =
46    Box<dyn Fn(Ctx<'js>, ReadableStreamControllerClass<'js>) -> Result<Promise<'js>> + 'js>;
47
48#[allow(private_interfaces)]
49#[derive(Clone)]
50pub enum PullAlgorithm<'js> {
51    ReturnPromiseUndefined,
52    Function {
53        f: Function<'js>,
54        underlying_source: Null<Undefined<Object<'js>>>,
55    },
56    RustFunction(Rc<PullRustFn<'js>>),
57    Tee(Class<'js, TeeState<'js>>),
58    Transform(TransformStreamClass<'js>),
59}
60
61impl<'js> Trace<'js> for PullAlgorithm<'js> {
62    fn trace<'a>(&self, tracer: rquickjs::class::Tracer<'a, 'js>) {
63        match self {
64            Self::ReturnPromiseUndefined => {},
65            Self::Function {
66                f,
67                underlying_source,
68            } => {
69                f.trace(tracer);
70                underlying_source.trace(tracer);
71            },
72            Self::RustFunction(_) => {},
73            Self::Tee(state) => state.trace(tracer),
74            Self::Transform(stream) => stream.trace(tracer),
75        }
76    }
77}
78
79unsafe impl<'js> JsLifetime<'js> for PullAlgorithm<'js> {
80    type Changed<'to> = PullAlgorithm<'to>;
81}
82
83impl<'js> PullAlgorithm<'js> {
84    pub fn from_fn(
85        f: impl Fn(Ctx<'js>, ReadableStreamControllerClass<'js>) -> Result<Promise<'js>> + 'js,
86    ) -> Self {
87        Self::RustFunction(Rc::new(Box::new(f)))
88    }
89
90    /// Wrap a one-shot pull closure. Subsequent invocations after the first
91    /// resolve with `undefined` without calling `f` again — useful for
92    /// streams that enqueue their whole payload in one go and then close.
93    pub fn from_fn_once(
94        f: impl FnOnce(Ctx<'js>, ReadableStreamControllerClass<'js>) -> Result<Promise<'js>> + 'js,
95    ) -> Self {
96        type OnceSlot<'js> = Rc<
97            RefCell<
98                Option<
99                    Box<
100                        dyn FnOnce(
101                                Ctx<'js>,
102                                ReadableStreamControllerClass<'js>,
103                            ) -> Result<Promise<'js>>
104                            + 'js,
105                    >,
106                >,
107            >,
108        >;
109        let slot: OnceSlot<'js> = Rc::new(RefCell::new(Some(Box::new(f))));
110        Self::from_fn(move |ctx, ctrl| {
111            if let Some(f) = slot.borrow_mut().take() {
112                f(ctx, ctrl)
113            } else {
114                Ok(PromisePrimordials::get(&ctx)?
115                    .promise_resolved_with_undefined
116                    .clone())
117            }
118        })
119    }
120
121    pub(super) fn from_tee_state(state: Class<'js, TeeState<'js>>) -> Self {
122        Self::Tee(state)
123    }
124
125    pub(crate) fn call(
126        &self,
127        ctx: Ctx<'js>,
128        promise_primordials: &PromisePrimordials<'js>,
129        controller: ReadableStreamControllerClass<'js>,
130    ) -> Result<Promise<'js>> {
131        match self {
132            PullAlgorithm::ReturnPromiseUndefined => {
133                Ok(promise_primordials.promise_resolved_with_undefined.clone())
134            },
135            PullAlgorithm::Function {
136                f,
137                underlying_source,
138            } => promise_resolved_with(
139                &ctx,
140                promise_primordials,
141                f.call::<_, Value>((This(underlying_source.clone()), controller)),
142            ),
143            PullAlgorithm::RustFunction(f) => f(ctx, controller),
144            PullAlgorithm::Tee(state) => {
145                crate::stream_web::readable::stream::tee::tee_pull_algorithm(ctx, state.clone())
146            },
147            PullAlgorithm::Transform(stream) => {
148                transform_stream::source_pull_algorithm(ctx, stream)
149            },
150        }
151    }
152}
153
154type CancelRustFn<'js> = Box<dyn FnOnce(Value<'js>) -> Result<Promise<'js>> + 'js>;
155
156#[allow(private_interfaces)]
157pub enum CancelAlgorithm<'js> {
158    ReturnPromiseUndefined,
159    Function {
160        f: Function<'js>,
161        underlying_source: Null<Undefined<Object<'js>>>,
162    },
163    RustFunction(Rc<RefCell<Option<CancelRustFn<'js>>>>),
164    Tee1(Class<'js, TeeState<'js>>),
165    Tee2(Class<'js, TeeState<'js>>),
166    Transform {
167        stream: TransformStreamClass<'js>,
168        controller: TransformStreamDefaultControllerClass<'js>,
169    },
170}
171
172impl<'js> Clone for CancelAlgorithm<'js> {
173    fn clone(&self) -> Self {
174        match self {
175            Self::ReturnPromiseUndefined => Self::ReturnPromiseUndefined,
176            Self::Function {
177                f,
178                underlying_source,
179            } => Self::Function {
180                f: f.clone(),
181                underlying_source: underlying_source.clone(),
182            },
183            Self::RustFunction(rc) => Self::RustFunction(rc.clone()),
184            Self::Tee1(state) => Self::Tee1(state.clone()),
185            Self::Tee2(state) => Self::Tee2(state.clone()),
186            Self::Transform { stream, controller } => Self::Transform {
187                stream: stream.clone(),
188                controller: controller.clone(),
189            },
190        }
191    }
192}
193
194impl<'js> Trace<'js> for CancelAlgorithm<'js> {
195    fn trace<'a>(&self, tracer: rquickjs::class::Tracer<'a, 'js>) {
196        match self {
197            Self::ReturnPromiseUndefined => {},
198            Self::Function {
199                f,
200                underlying_source,
201            } => {
202                f.trace(tracer);
203                underlying_source.trace(tracer);
204            },
205            Self::RustFunction(_) => {},
206            Self::Tee1(state) | Self::Tee2(state) => state.trace(tracer),
207            Self::Transform { stream, controller } => {
208                stream.trace(tracer);
209                controller.trace(tracer);
210            },
211        }
212    }
213}
214
215unsafe impl<'js> JsLifetime<'js> for CancelAlgorithm<'js> {
216    type Changed<'to> = CancelAlgorithm<'to>;
217}
218
219impl<'js> CancelAlgorithm<'js> {
220    pub fn from_fn(f: impl FnOnce(Value<'js>) -> Result<Promise<'js>> + 'js) -> Self {
221        Self::RustFunction(Rc::new(RefCell::new(Some(Box::new(f)))))
222    }
223
224    pub(super) fn from_tee_state_1(state: Class<'js, TeeState<'js>>) -> Self {
225        Self::Tee1(state)
226    }
227
228    pub(super) fn from_tee_state_2(state: Class<'js, TeeState<'js>>) -> Self {
229        Self::Tee2(state)
230    }
231
232    pub(crate) fn call(
233        &self,
234        ctx: Ctx<'js>,
235        promise_primordials: &PromisePrimordials<'js>,
236        reason: Value<'js>,
237    ) -> Result<Promise<'js>> {
238        match self {
239            CancelAlgorithm::ReturnPromiseUndefined => {
240                Ok(promise_primordials.promise_resolved_with_undefined.clone())
241            },
242            CancelAlgorithm::Function {
243                f,
244                underlying_source,
245            } => {
246                let result: Result<Value> = f.call((This(underlying_source.clone()), reason));
247                promise_resolved_with(&ctx, promise_primordials, result)
248            },
249            CancelAlgorithm::RustFunction(f) => {
250                let f = f
251                    .borrow_mut()
252                    .take()
253                    .expect("cancel algorithm must only be called once");
254                f(reason)
255            },
256            CancelAlgorithm::Tee1(state) => {
257                crate::stream_web::readable::stream::tee::tee_cancel_algorithm(ctx, state.clone(), reason, 0)
258            },
259            CancelAlgorithm::Tee2(state) => {
260                crate::stream_web::readable::stream::tee::tee_cancel_algorithm(ctx, state.clone(), reason, 1)
261            },
262            CancelAlgorithm::Transform { stream, controller } => {
263                transform_stream::source_cancel_algorithm(ctx, stream, controller, reason)
264            },
265        }
266    }
267}
ferrijs_std/stream_web/readable/stream/algorithms.rs

ferrijs_std/stream_web/readable/stream/
algorithms.rs
