ferrijs_std/web/
compression.rs

1//! WHATWG `CompressionStream` / `DecompressionStream`.
2//!
3//! Both are "generic transform streams": not `TransformStream`
4//! subclasses, but objects exposing the `readable` / `writable` pair of
5//! one. This builds a real `TransformStream` (the vendored class in
6//! [`crate::stream_web`]) whose `transform` and `flush` are
7//! native functions, so backpressure, cancellation and `pipeThrough`
8//! all come from the spec-exact stream machinery rather than being
9//! reimplemented here.
10//!
11//! The spec defines exactly three formats — `gzip`, `deflate` (zlib
12//! wrapper) and `deflate-raw` — all of which `flate2` covers. Brotli and
13//! zstd are deliberately absent: they are not in the Compression Streams
14//! spec, and accepting them would be a silent extension callers could
15//! not rely on elsewhere.
16//!
17//! The native closures capture only an `Arc<Mutex<..>>` of Rust state,
18//! never a JS value, per the GC-cycle discipline.
19
20use std::io::Write as _;
21use std::sync::{Arc, Mutex};
22
23use rquickjs::function::Func;
24use rquickjs::{Class, Ctx, Function, Object, TypedArray, Value, class::Trace};
25
26/// The streaming coder behind one `CompressionStream` /
27/// `DecompressionStream`. Each variant writes into a `Vec<u8>` that is
28/// drained after every chunk, so nothing accumulates beyond one
29/// transform step.
30enum Coder {
31  GzipEncode(flate2::write::GzEncoder<Vec<u8>>),
32  DeflateEncode(flate2::write::ZlibEncoder<Vec<u8>>),
33  DeflateRawEncode(flate2::write::DeflateEncoder<Vec<u8>>),
34  GzipDecode(flate2::write::GzDecoder<Vec<u8>>),
35  DeflateDecode(flate2::write::ZlibDecoder<Vec<u8>>),
36  DeflateRawDecode(flate2::write::DeflateDecoder<Vec<u8>>),
37}
38
39impl Coder {
40  fn new(format: &str, decompress: bool) -> Option<Self> {
41    let level = flate2::Compression::default();
42    Some(match (format, decompress) {
43      ("gzip", false) => Self::GzipEncode(flate2::write::GzEncoder::new(Vec::new(), level)),
44      ("deflate", false) => Self::DeflateEncode(flate2::write::ZlibEncoder::new(Vec::new(), level)),
45      ("deflate-raw", false) => Self::DeflateRawEncode(flate2::write::DeflateEncoder::new(Vec::new(), level)),
46      ("gzip", true) => Self::GzipDecode(flate2::write::GzDecoder::new(Vec::new())),
47      ("deflate", true) => Self::DeflateDecode(flate2::write::ZlibDecoder::new(Vec::new())),
48      ("deflate-raw", true) => Self::DeflateRawDecode(flate2::write::DeflateDecoder::new(Vec::new())),
49      _ => return None,
50    })
51  }
52
53  /// Feed input and return whatever output became available. Returning
54  /// an empty vec is normal — a coder may buffer internally until it has
55  /// a full block.
56  fn push(&mut self, data: &[u8]) -> std::io::Result<Vec<u8>> {
57    match self {
58      Self::GzipEncode(c) => {
59        c.write_all(data)?;
60        Ok(std::mem::take(c.get_mut()))
61      },
62      Self::DeflateEncode(c) => {
63        c.write_all(data)?;
64        Ok(std::mem::take(c.get_mut()))
65      },
66      Self::DeflateRawEncode(c) => {
67        c.write_all(data)?;
68        Ok(std::mem::take(c.get_mut()))
69      },
70      Self::GzipDecode(c) => {
71        c.write_all(data)?;
72        Ok(std::mem::take(c.get_mut()))
73      },
74      Self::DeflateDecode(c) => {
75        c.write_all(data)?;
76        Ok(std::mem::take(c.get_mut()))
77      },
78      Self::DeflateRawDecode(c) => {
79        c.write_all(data)?;
80        Ok(std::mem::take(c.get_mut()))
81      },
82    }
83  }
84
85  /// End the stream and return the trailing output (gzip's CRC/length
86  /// trailer, the deflate final block, …).
87  fn finish(self) -> std::io::Result<Vec<u8>> {
88    match self {
89      Self::GzipEncode(c) => c.finish(),
90      Self::DeflateEncode(c) => c.finish(),
91      Self::DeflateRawEncode(c) => c.finish(),
92      Self::GzipDecode(c) => c.finish(),
93      Self::DeflateDecode(c) => c.finish(),
94      Self::DeflateRawDecode(c) => c.finish(),
95    }
96  }
97}
98
99/// `Some` until `flush` consumes it; `None` after, so a late write
100/// cannot resurrect a finished coder.
101type SharedCoder = Arc<Mutex<Option<Coder>>>;
102
103fn lock(coder: &SharedCoder) -> std::sync::MutexGuard<'_, Option<Coder>> {
104  coder.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
105}
106
107/// Bytes of a `BufferSource` chunk. Per spec anything else is a
108/// `TypeError` — a string is NOT encoded implicitly, because the caller
109/// would silently get UTF-8 where they may have meant something else.
110fn buffer_source_bytes<'js>(ctx: &Ctx<'js>, chunk: &Value<'js>) -> rquickjs::Result<Vec<u8>> {
111  // The shared extractor, with this call site's spec wording on failure.
112  crate::node::bytes::buffer_source_bytes(ctx, chunk).map_err(|_| {
113    rquickjs::Exception::throw_type(
114      ctx,
115      "Failed to execute 'write': chunk could not be converted to a BufferSource",
116    )
117  })
118}
119
120/// Hand `bytes` to the transform controller, skipping an empty step (the
121/// spec enqueues only when the coder actually produced output).
122fn enqueue(controller: &Object<'_>, bytes: Vec<u8>) -> rquickjs::Result<()> {
123  if bytes.is_empty() {
124    return Ok(());
125  }
126  let chunk = TypedArray::new(controller.ctx().clone(), bytes)?.into_value();
127  controller
128    .get::<_, Function<'_>>("enqueue")?
129    .call::<_, ()>((rquickjs::function::This(controller.clone()), chunk))
130}
131
132fn io_error(ctx: &Ctx<'_>, what: &str, e: &std::io::Error) -> rquickjs::Error {
133  rquickjs::Exception::throw_type(ctx, &format!("{what}: {e}"))
134}
135
136/// Build the `TransformStream` that backs a generic transform stream,
137/// with native `transform` / `flush` driving `coder`.
138fn transform_stream<'js>(ctx: &Ctx<'js>, coder: SharedCoder) -> rquickjs::Result<Object<'js>> {
139  let transformer = Object::new(ctx.clone())?;
140  {
141    let coder = coder.clone();
142    transformer.set(
143      "transform",
144      Func::from(
145        move |ctx: Ctx<'js>, chunk: Value<'js>, controller: Object<'js>| -> rquickjs::Result<()> {
146          let bytes = buffer_source_bytes(&ctx, &chunk)?;
147          let out = {
148            let mut guard = lock(&coder);
149            let Some(coder) = guard.as_mut() else {
150              return Err(rquickjs::Exception::throw_type(&ctx, "the stream is already closed"));
151            };
152            coder
153              .push(&bytes)
154              .map_err(|e| io_error(&ctx, "compression failed", &e))?
155          };
156          enqueue(&controller, out)
157        },
158      ),
159    )?;
160  }
161  {
162    let coder = coder.clone();
163    transformer.set(
164      "flush",
165      Func::from(move |ctx: Ctx<'js>, controller: Object<'js>| -> rquickjs::Result<()> {
166        let out = match lock(&coder).take() {
167          None => return Ok(()),
168          Some(coder) => coder
169            .finish()
170            .map_err(|e| io_error(&ctx, "compression failed at end of stream", &e))?,
171        };
172        enqueue(&controller, out)
173      }),
174    )?;
175  }
176
177  ctx
178    .globals()
179    .get::<_, rquickjs::function::Constructor<'js>>("TransformStream")?
180    .construct((transformer,))
181}
182
183/// The `readable` / `writable` pair every generic transform stream
184/// exposes. Both classes below are this plus a constructor.
185#[derive(Trace)]
186struct Duplex<'js> {
187  readable: Value<'js>,
188  writable: Value<'js>,
189}
190
191impl<'js> Duplex<'js> {
192  fn new(ctx: &Ctx<'js>, format: &str, decompress: bool) -> rquickjs::Result<Self> {
193    let what = if decompress {
194      "DecompressionStream"
195    } else {
196      "CompressionStream"
197    };
198    let Some(coder) = Coder::new(format, decompress) else {
199      return Err(rquickjs::Exception::throw_type(
200        ctx,
201        &format!(
202          "Failed to construct '{what}': '{format}' is not a valid enum value of type CompressionFormat \
203           (expected 'gzip', 'deflate' or 'deflate-raw')"
204        ),
205      ));
206    };
207    let stream = transform_stream(ctx, Arc::new(Mutex::new(Some(coder))))?;
208    Ok(Self {
209      readable: stream.get("readable")?,
210      writable: stream.get("writable")?,
211    })
212  }
213}
214
215/// WHATWG `CompressionStream`.
216#[derive(Trace)]
217#[rquickjs::class(rename = "CompressionStream")]
218pub struct CompressionStreamJs<'js> {
219  inner: Duplex<'js>,
220}
221
222/// WHATWG `DecompressionStream`.
223#[derive(Trace)]
224#[rquickjs::class(rename = "DecompressionStream")]
225pub struct DecompressionStreamJs<'js> {
226  inner: Duplex<'js>,
227}
228
229#[allow(unsafe_code)]
230unsafe impl<'js> rquickjs::JsLifetime<'js> for CompressionStreamJs<'js> {
231  type Changed<'to> = CompressionStreamJs<'to>;
232}
233#[allow(unsafe_code)]
234unsafe impl<'js> rquickjs::JsLifetime<'js> for DecompressionStreamJs<'js> {
235  type Changed<'to> = DecompressionStreamJs<'to>;
236}
237
238#[rquickjs::methods]
239impl<'js> CompressionStreamJs<'js> {
240  /// Spec: every platform object carries `Symbol.toStringTag`, so
241  /// `Object.prototype.toString.call(x)` reads `[object CompressionStream]`.
242  #[qjs(prop, rename = rquickjs::atom::PredefinedAtom::SymbolToStringTag, configurable)]
243  pub fn to_string_tag() -> &'static str {
244    "CompressionStream"
245  }
246
247  #[qjs(constructor)]
248  pub fn new(ctx: Ctx<'js>, format: String) -> rquickjs::Result<Self> {
249    Ok(Self {
250      inner: Duplex::new(&ctx, &format, false)?,
251    })
252  }
253
254  #[qjs(get, rename = "readable")]
255  pub fn readable(&self) -> Value<'js> {
256    self.inner.readable.clone()
257  }
258
259  #[qjs(get, rename = "writable")]
260  pub fn writable(&self) -> Value<'js> {
261    self.inner.writable.clone()
262  }
263}
264
265#[rquickjs::methods]
266impl<'js> DecompressionStreamJs<'js> {
267  /// Spec: every platform object carries `Symbol.toStringTag`, so
268  /// `Object.prototype.toString.call(x)` reads `[object DecompressionStream]`.
269  #[qjs(prop, rename = rquickjs::atom::PredefinedAtom::SymbolToStringTag, configurable)]
270  pub fn to_string_tag() -> &'static str {
271    "DecompressionStream"
272  }
273
274  #[qjs(constructor)]
275  pub fn new(ctx: Ctx<'js>, format: String) -> rquickjs::Result<Self> {
276    Ok(Self {
277      inner: Duplex::new(&ctx, &format, true)?,
278    })
279  }
280
281  #[qjs(get, rename = "readable")]
282  pub fn readable(&self) -> Value<'js> {
283    self.inner.readable.clone()
284  }
285
286  #[qjs(get, rename = "writable")]
287  pub fn writable(&self) -> Value<'js> {
288    self.inner.writable.clone()
289  }
290}
291
292pub fn install(ctx: &Ctx<'_>) -> rquickjs::Result<()> {
293  let globals = ctx.globals();
294  Class::<CompressionStreamJs<'_>>::define(&globals)?;
295  Class::<DecompressionStreamJs<'_>>::define(&globals)?;
296  Ok(())
297}
ferrijs_std/web/compression.rs

ferrijs_std/web/
compression.rs
