Stream implement pipeThrough (#36977)

Part of https://github.com/servo/servo/issues/34676

https://github.com/servo/servo/pull/36905 needs to be merged first.

---------

Signed-off-by: Taym Haddadi <haddadi.taym@gmail.com>

components/script/dom/bindings/structuredclone.rs

@@ -14,7 +14,7 @@ use base::id::{
 };
 use constellation_traits::{
     BlobImpl, DomException, DomPoint, MessagePortImpl, Serializable as SerializableInterface,
-    StructuredSerializedData, Transferrable as TransferrableInterface,
+    StructuredSerializedData, Transferrable as TransferrableInterface, TransformStreamData,
 };
 use js::gc::RootedVec;
 use js::glue::{
@@ -517,6 +517,8 @@ pub(crate) struct StructuredDataReader<'a> {
     /// used as part of the "transfer-receiving" steps of ports,
     /// to produce the DOM ports stored in `message_ports` above.
     pub(crate) port_impls: Option<HashMap<MessagePortId, MessagePortImpl>>,
+    /// A map of transform stream implementations,
+    pub(crate) transform_streams_port_impls: Option<HashMap<MessagePortId, TransformStreamData>>,
     /// A map of blob implementations,
     /// used as part of the "deserialize" steps of blobs,
     /// to produce the DOM blobs stored in `blobs` above.
@@ -535,6 +537,8 @@ pub(crate) struct StructuredDataWriter {
     pub(crate) errors: DOMErrorRecord,
     /// Transferred ports.
     pub(crate) ports: Option<HashMap<MessagePortId, MessagePortImpl>>,
+    /// Transferred transform streams.
+    pub(crate) transform_streams_port: Option<HashMap<MessagePortId, TransformStreamData>>,
     /// Serialized points.
     pub(crate) points: Option<HashMap<DomPointId, DomPoint>>,
     /// Serialized exceptions.
@@ -591,6 +595,7 @@ pub(crate) fn write(
         let data = StructuredSerializedData {
             serialized: data,
             ports: sc_writer.ports.take(),
+            transform_streams: sc_writer.transform_streams_port.take(),
             points: sc_writer.points.take(),
             exceptions: sc_writer.exceptions.take(),
             blobs: sc_writer.blobs.take(),
@@ -613,6 +618,7 @@ pub(crate) fn read(
     let mut sc_reader = StructuredDataReader {
         roots,
         port_impls: data.ports.take(),
+        transform_streams_port_impls: data.transform_streams.take(),
         blob_impls: data.blobs.take(),
         points: data.points.take(),
         exceptions: data.exceptions.take(),

components/script/dom/readablestream.rs

@@ -24,7 +24,7 @@ use js::typedarray::ArrayBufferViewU8;
 use crate::dom::bindings::codegen::Bindings::QueuingStrategyBinding::QueuingStrategy;
 use crate::dom::bindings::codegen::Bindings::ReadableStreamBinding::{
     ReadableStreamGetReaderOptions, ReadableStreamMethods, ReadableStreamReaderMode,
-    StreamPipeOptions,
+    ReadableWritablePair, StreamPipeOptions,
 };
 use script_bindings::str::DOMString;
 
@@ -2006,6 +2006,50 @@ impl ReadableStreamMethods<crate::DomTypeHolder> for ReadableStream {
             can_gc,
         )
     }
+
+    /// <https://streams.spec.whatwg.org/#rs-pipe-through>
+    fn PipeThrough(
+        &self,
+        transform: &ReadableWritablePair,
+        options: &StreamPipeOptions,
+        realm: InRealm,
+        can_gc: CanGc,
+    ) -> Fallible<DomRoot<ReadableStream>> {
+        let global = self.global();
+        let cx = GlobalScope::get_cx();
+
+        // If ! IsReadableStreamLocked(this) is true, throw a TypeError exception.
+        if self.is_locked() {
+            return Err(Error::Type("Source stream is locked".to_owned()));
+        }
+
+        // If ! IsWritableStreamLocked(transform["writable"]) is true, throw a TypeError exception.
+        if transform.writable.is_locked() {
+            return Err(Error::Type("Destination stream is locked".to_owned()));
+        }
+
+        // Let signal be options["signal"] if it exists, or undefined otherwise.
+        // TODO: implement AbortSignal.
+
+        // Let promise be ! ReadableStreamPipeTo(this, transform["writable"],
+        // options["preventClose"], options["preventAbort"], options["preventCancel"], signal).
+        let promise = self.pipe_to(
+            cx,
+            &global,
+            &transform.writable,
+            options.preventAbort,
+            options.preventCancel,
+            options.preventClose,
+            realm,
+            can_gc,
+        );
+
+        // Set promise.[[PromiseIsHandled]] to true.
+        promise.set_promise_is_handled();
+
+        // Return transform["readable"].
+        Ok(transform.readable.clone())
+    }
 }
 
 #[allow(unsafe_code)]

components/script/dom/transformstream.rs

@@ -8,7 +8,7 @@ use std::ptr::{self};
 use std::rc::Rc;
 
 use base::id::{MessagePortId, MessagePortIndex};
-use constellation_traits::MessagePortImpl;
+use constellation_traits::TransformStreamData;
 use dom_struct::dom_struct;
 use js::jsapi::{Heap, IsPromiseObject, JSObject};
 use js::jsval::{JSVal, ObjectValue, UndefinedValue};
@@ -1007,9 +1007,9 @@ impl TransformStreamMethods<crate::DomTypeHolder> for TransformStream {
 /// <https://streams.spec.whatwg.org/#ts-transfer>
 impl Transferable for TransformStream {
     type Index = MessagePortIndex;
-    type Data = MessagePortImpl;
+    type Data = TransformStreamData;
 
-    fn transfer(&self) -> Result<(MessagePortId, MessagePortImpl), ()> {
+    fn transfer(&self) -> Result<(MessagePortId, TransformStreamData), ()> {
         let global = self.global();
         let realm = enter_realm(&*global);
         let comp = InRealm::Entered(&realm);
@@ -1023,73 +1023,85 @@ impl Transferable for TransformStream {
         let writable = self.get_writable();
 
         // If ! IsReadableStreamLocked(readable) is true, throw a "DataCloneError" DOMException.
-        if readable.is_locked() {
-            return Err(());
-        }
-
         // If ! IsWritableStreamLocked(writable) is true, throw a "DataCloneError" DOMException.
-        if writable.is_locked() {
+        if readable.is_locked() || writable.is_locked() {
             return Err(());
         }
 
-        // Create the shared port pair
-        let port_1 = MessagePort::new(&global, can_gc);
-        global.track_message_port(&port_1, None);
-        let port_2 = MessagePort::new(&global, can_gc);
-        global.track_message_port(&port_2, None);
-        global.entangle_ports(*port_1.message_port_id(), *port_2.message_port_id());
+        // First port pair (readable → proxy writable)
+        let port1 = MessagePort::new(&global, can_gc);
+        global.track_message_port(&port1, None);
+        let port1_peer = MessagePort::new(&global, can_gc);
+        global.track_message_port(&port1_peer, None);
+        global.entangle_ports(*port1.message_port_id(), *port1_peer.message_port_id());
+
+        let proxy_readable = ReadableStream::new_with_proto(&global, None, can_gc);
+        proxy_readable.setup_cross_realm_transform_readable(cx, &port1, can_gc);
+        proxy_readable
+            .pipe_to(cx, &global, &writable, false, false, false, comp, can_gc)
+            .set_promise_is_handled();
+
+        // Second port pair (proxy readable → writable)
+        let port2 = MessagePort::new(&global, can_gc);
+        global.track_message_port(&port2, None);
+        let port2_peer = MessagePort::new(&global, can_gc);
+        global.track_message_port(&port2_peer, None);
+        global.entangle_ports(*port2.message_port_id(), *port2_peer.message_port_id());
 
-        // Create a proxy WritableStream wired to port_1
         let proxy_writable = WritableStream::new_with_proto(&global, None, can_gc);
-        proxy_writable.setup_cross_realm_transform_writable(cx, &port_1, can_gc);
+        proxy_writable.setup_cross_realm_transform_writable(cx, &port2, can_gc);
 
         // Pipe readable into the proxy writable (→ port_1)
-        let pipe1 = readable.pipe_to(
-            cx,
-            &global,
-            &proxy_writable,
-            false,
-            false,
-            false,
-            comp,
-            can_gc,
-        );
-        pipe1.set_promise_is_handled();
-
-        // Create a proxy ReadableStream wired to port_1
-        let proxy_readable = ReadableStream::new_with_proto(&global, None, can_gc);
-        proxy_readable.setup_cross_realm_transform_readable(cx, &port_1, can_gc);
-
-        // Pipe proxy readable (← port_1) into writable
-        let pipe2 =
-            proxy_readable.pipe_to(cx, &global, &writable, false, false, false, comp, can_gc);
-        pipe2.set_promise_is_handled();
+        readable
+            .pipe_to(
+                cx,
+                &global,
+                &proxy_writable,
+                false,
+                false,
+                false,
+                comp,
+                can_gc,
+            )
+            .set_promise_is_handled();
 
         // Set dataHolder.[[readable]] to ! StructuredSerializeWithTransfer(readable, « readable »).
         // Set dataHolder.[[writable]] to ! StructuredSerializeWithTransfer(writable, « writable »).
-        port_2.transfer()
+        Ok((
+            *port1_peer.message_port_id(),
+            TransformStreamData {
+                readable: port1_peer.transfer()?,
+                writable: port2_peer.transfer()?,
+            },
+        ))
     }
 
     fn transfer_receive(
         owner: &GlobalScope,
-        id: MessagePortId,
-        port_impl: MessagePortImpl,
+        _id: MessagePortId,
+        data: TransformStreamData,
     ) -> Result<DomRoot<Self>, ()> {
         let can_gc = CanGc::note();
+        let cx = GlobalScope::get_cx();
+
+        let port1 = MessagePort::transfer_receive(owner, data.readable.0, data.readable.1)?;
+        let port2 = MessagePort::transfer_receive(owner, data.writable.0, data.writable.1)?;
 
         // Let readableRecord be ! StructuredDeserializeWithTransfer(dataHolder.[[readable]], the current Realm).
         // Set value.[[readable]] to readableRecord.[[Deserialized]].
-        let readable = ReadableStream::transfer_receive(owner, id, port_impl.clone())?;
-
         // Let writableRecord be ! StructuredDeserializeWithTransfer(dataHolder.[[writable]], the current Realm).
-        let writable = WritableStream::transfer_receive(owner, id, port_impl)?;
+        let proxy_readable = ReadableStream::new_with_proto(owner, None, can_gc);
+        proxy_readable.setup_cross_realm_transform_readable(cx, &port2, can_gc);
+
+        let proxy_writable = WritableStream::new_with_proto(owner, None, can_gc);
+        proxy_writable.setup_cross_realm_transform_writable(cx, &port1, can_gc);
 
         // Set value.[[readable]] to readableRecord.[[Deserialized]].
         // Set value.[[writable]] to writableRecord.[[Deserialized]].
         // Set value.[[backpressure]], value.[[backpressureChangePromise]], and value.[[controller]] to undefined.
         let stream = TransformStream::new_with_proto(owner, None, can_gc);
-        stream.readable.set(Some(&readable));
-        stream.writable.set(Some(&writable));
+        stream.readable.set(Some(&proxy_readable));
+        stream.writable.set(Some(&proxy_writable));
 
         Ok(stream)
     }
@@ -1098,8 +1110,8 @@ impl Transferable for TransformStream {
         data: StructuredData<'a, '_>,
     ) -> &'a mut Option<HashMap<MessagePortId, Self::Data>> {
         match data {
-            StructuredData::Reader(r) => &mut r.port_impls,
-            StructuredData::Writer(w) => &mut w.ports,
+            StructuredData::Reader(r) => &mut r.transform_streams_port_impls,
+            StructuredData::Writer(w) => &mut w.transform_streams_port,
         }
     }
 }

components/script_bindings/codegen/Bindings.conf

@@ -721,8 +721,8 @@ DOMInterfaces = {
 },
 
 'ReadableStream': {
-    'canGc': ['GetReader', 'Cancel', 'PipeTo', 'Tee'],
-    'inRealms': ['PipeTo'],
+    'canGc': ['GetReader', 'Cancel', 'PipeTo', 'PipeThrough', 'Tee'],
+    'inRealms': ['PipeTo', 'PipeThrough'],
 },
 
 "ReadableStreamDefaultController": {

components/script_bindings/webidls/ReadableStream.webidl

@@ -20,8 +20,8 @@ interface _ReadableStream {
   [Throws]
   ReadableStreamReader getReader(optional ReadableStreamGetReaderOptions options = {});
 
-  // [Throws]
-  // ReadableStream pipeThrough(ReadableWritablePair transform, optional StreamPipeOptions options = {});
+  [Throws]
+  ReadableStream pipeThrough(ReadableWritablePair transform, optional StreamPipeOptions options = {});
 
   [NewObject]
   Promise<undefined> pipeTo(WritableStream destination, optional StreamPipeOptions options = {});

components/shared/constellation/lib.rs

@@ -152,7 +152,7 @@ pub enum TraversalDirection {
 }
 
 /// A task on the <https://html.spec.whatwg.org/multipage/#port-message-queue>
-#[derive(Clone, Debug, Deserialize, MallocSizeOf, Serialize)]
+#[derive(Debug, Deserialize, MallocSizeOf, Serialize)]
 pub struct PortMessageTask {
     /// The origin of this task.
     pub origin: ImmutableOrigin,

components/shared/constellation/structured_data/mod.rs

@@ -20,7 +20,7 @@ pub use transferable::*;
 
 /// A data-holder for serialized data and transferred objects.
 /// <https://html.spec.whatwg.org/multipage/#structuredserializewithtransfer>
-#[derive(Clone, Debug, Default, Deserialize, MallocSizeOf, Serialize)]
+#[derive(Debug, Default, Deserialize, MallocSizeOf, Serialize)]
 pub struct StructuredSerializedData {
     /// Data serialized by SpiderMonkey.
     pub serialized: Vec<u8>,
@@ -32,6 +32,8 @@ pub struct StructuredSerializedData {
     pub exceptions: Option<HashMap<DomExceptionId, DomException>>,
     /// Transferred objects.
     pub ports: Option<HashMap<MessagePortId, MessagePortImpl>>,
+    /// Transform streams transferred objects.
+    pub transform_streams: Option<HashMap<MessagePortId, TransformStreamData>>,
 }
 
 impl StructuredSerializedData {

components/shared/constellation/structured_data/serializable.rs

@@ -88,7 +88,7 @@ impl Clone for BroadcastMsg {
 }
 
 /// File-based blob
-#[derive(Clone, Debug, Deserialize, MallocSizeOf, Serialize)]
+#[derive(Debug, Deserialize, MallocSizeOf, Serialize)]
 pub struct FileBlob {
     #[ignore_malloc_size_of = "Uuid are hard(not really)"]
     id: Uuid,
@@ -164,7 +164,7 @@ impl BroadcastClone for BlobImpl {
 }
 
 /// The data backing a DOM Blob.
-#[derive(Clone, Debug, Deserialize, MallocSizeOf, Serialize)]
+#[derive(Debug, Deserialize, MallocSizeOf, Serialize)]
 pub struct BlobImpl {
     /// UUID of the blob.
     blob_id: BlobId,
@@ -177,7 +177,7 @@ pub struct BlobImpl {
 }
 
 /// Different backends of Blob
-#[derive(Clone, Debug, Deserialize, MallocSizeOf, Serialize)]
+#[derive(Debug, Deserialize, MallocSizeOf, Serialize)]
 pub enum BlobData {
     /// File-based blob, whose content lives in the net process
     File(FileBlob),

components/shared/constellation/structured_data/transferable.rs

@@ -15,6 +15,12 @@ use strum::EnumIter;
 
 use crate::PortMessageTask;
 
+#[derive(Debug, Deserialize, MallocSizeOf, Serialize)]
+pub struct TransformStreamData {
+    pub readable: (MessagePortId, MessagePortImpl),
+    pub writable: (MessagePortId, MessagePortImpl),
+}
+
 /// All the DOM interfaces that can be transferred.
 #[derive(Clone, Copy, Debug, EnumIter)]
 pub enum Transferrable {
@@ -28,7 +34,7 @@ pub enum Transferrable {
     TransformStream,
 }
 
-#[derive(Clone, Debug, Deserialize, MallocSizeOf, Serialize)]
+#[derive(Debug, Deserialize, MallocSizeOf, Serialize)]
 enum MessagePortState {
     /// <https://html.spec.whatwg.org/multipage/#detached>
     Detached,
@@ -42,7 +48,7 @@ enum MessagePortState {
     Disabled(bool),
 }
 
-#[derive(Clone, Debug, Deserialize, MallocSizeOf, Serialize)]
+#[derive(Debug, Deserialize, MallocSizeOf, Serialize)]
 /// The data and logic backing the DOM managed MessagePort.
 pub struct MessagePortImpl {
     /// The current state of the port.

tests/wpt/meta/fetch/api/response/response-stream-disturbed-by-pipe.any.js.ini

@@ -1,13 +1,3 @@
-[response-stream-disturbed-by-pipe.any.worker.html]
-  [using pipeThrough on Response body should disturb it synchronously]
-    expected: FAIL
-
-
-[response-stream-disturbed-by-pipe.any.html]
-  [using pipeThrough on Response body should disturb it synchronously]
-    expected: FAIL
-
-
 [response-stream-disturbed-by-pipe.any.serviceworker.html]
   expected: ERROR
 

tests/wpt/meta/streams/piping/pipe-through.any.js.ini

@@ -6,63 +6,6 @@
 
 [pipe-through.any.worker.html]
   expected: ERROR
-  [Piping through a duck-typed pass-through transform stream should work]
-    expected: FAIL
-
-  [Piping through a transform errored on the writable end does not cause an unhandled promise rejection]
-    expected: FAIL
-
-  [pipeThrough should not call pipeTo on this]
-    expected: FAIL
-
-  [pipeThrough should not call pipeTo on the ReadableStream prototype]
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow 'null']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow 'undefined']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow '0']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow 'NaN']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow 'true']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow 'ReadableStream']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow '[object ReadableStream\]']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow 'null']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow 'undefined']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow '0']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow 'NaN']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow 'true']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow 'WritableStream']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow '[object WritableStream\]']
-    expected: FAIL
-
-  [pipeThrough should rethrow errors from accessing readable or writable]
-    expected: FAIL
-
 
 [pipe-through.any.sharedworker.html]
   expected: ERROR
@@ -84,59 +27,3 @@
 
 [pipe-through.any.html]
   expected: ERROR
-  [Piping through a duck-typed pass-through transform stream should work]
-    expected: FAIL
-
-  [Piping through a transform errored on the writable end does not cause an unhandled promise rejection]
-    expected: FAIL
-
-  [pipeThrough should not call pipeTo on this]
-    expected: FAIL
-
-  [pipeThrough should not call pipeTo on the ReadableStream prototype]
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow 'null']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow 'undefined']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow '0']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow 'NaN']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow 'true']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow 'ReadableStream']
-    expected: FAIL
-
-  [pipeThrough should brand-check this and not allow '[object ReadableStream\]']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow 'null']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow 'undefined']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow '0']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow 'NaN']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow 'true']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow 'WritableStream']
-    expected: FAIL
-
-  [pipeThrough should brand-check writable and not allow '[object WritableStream\]']
-    expected: FAIL
-
-  [pipeThrough should rethrow errors from accessing readable or writable]
-    expected: FAIL

tests/wpt/meta/streams/piping/throwing-options.any.js.ini

@@ -21,15 +21,6 @@
 
 [throwing-options.any.html]
   expected: TIMEOUT
-  [pipeThrough should stop after getting preventAbort throws]
-    expected: FAIL
-
-  [pipeThrough should stop after getting preventCancel throws]
-    expected: FAIL
-
-  [pipeThrough should stop after getting preventClose throws]
-    expected: FAIL
-
   [pipeTo should stop after getting signal throws]
     expected: TIMEOUT
 
@@ -39,15 +30,6 @@
 
 [throwing-options.any.worker.html]
   expected: TIMEOUT
-  [pipeThrough should stop after getting preventAbort throws]
-    expected: FAIL
-
-  [pipeThrough should stop after getting preventCancel throws]
-    expected: FAIL
-
-  [pipeThrough should stop after getting preventClose throws]
-    expected: FAIL
-
   [pipeTo should stop after getting signal throws]
     expected: TIMEOUT
 

tests/wpt/meta/streams/piping/transform-streams.any.js.ini

@@ -10,18 +10,10 @@
 [transform-streams.any.shadowrealm-in-sharedworker.html]
   expected: ERROR
 
-[transform-streams.any.worker.html]
-  [Piping through an identity transform stream should close the destination when the source closes]
-    expected: FAIL
-
 
 [transform-streams.any.shadowrealm-in-shadowrealm.html]
   expected: ERROR
 
-[transform-streams.any.html]
-  [Piping through an identity transform stream should close the destination when the source closes]
-    expected: FAIL
-
 
 [transform-streams.any.shadowrealm-in-window.html]
   expected: ERROR

tests/wpt/meta/streams/readable-byte-streams/templated.any.js.ini

@@ -21,12 +21,3 @@
 
 [templated.any.shadowrealm-in-sharedworker.html]
   expected: ERROR
-
-[templated.any.worker.html]
-  [ReadableStream with byte source (empty): instances have the correct methods and properties]
-    expected: FAIL
-
-
-[templated.any.html]
-  [ReadableStream with byte source (empty): instances have the correct methods and properties]
-    expected: FAIL

tests/wpt/meta/streams/readable-streams/templated.any.js.ini

@@ -4,13 +4,6 @@
 [templated.any.serviceworker.html]
   expected: ERROR
 
-[templated.any.html]
-  [ReadableStream (empty): instances have the correct methods and properties]
-    expected: FAIL
-
-[templated.any.worker.html]
-  [ReadableStream (empty): instances have the correct methods and properties]
-    expected: FAIL
 
 [templated.any.shadowrealm.html]
   expected: TIMEOUT

tests/wpt/meta/streams/transferable/transform-stream.html.ini

@@ -1,3 +0,0 @@
-[transform-stream.html]
-  [piping through transferred transforms should work]
-    expected: FAIL