integrate readablestream with fetch and blob

components/script/dom/globalscope.rs

@@ -43,6 +43,7 @@ use crate::dom::paintworkletglobalscope::PaintWorkletGlobalScope;
 use crate::dom::performance::Performance;
 use crate::dom::performanceobserver::VALID_ENTRY_TYPES;
 use crate::dom::promise::Promise;
+use crate::dom::readablestream::{ExternalUnderlyingSource, ReadableStream};
 use crate::dom::serviceworker::ServiceWorker;
 use crate::dom::serviceworkerregistration::ServiceWorkerRegistration;
 use crate::dom::window::Window;
@@ -319,11 +320,19 @@ struct FileListener {
     task_canceller: TaskCanceller,
 }
 
-struct FileListenerCallback(Box<dyn Fn(Rc<Promise>, Result<Vec<u8>, Error>) + Send>);
+enum FileListenerCallback {
+    Promise(Box<dyn Fn(Rc<Promise>, Result<Vec<u8>, Error>) + Send>),
+    Stream,
+}
+
+enum FileListenerTarget {
+    Promise(TrustedPromise),
+    Stream(Trusted<ReadableStream>),
+}
 
 enum FileListenerState {
-    Empty(FileListenerCallback, TrustedPromise),
-    Receiving(Vec<u8>, FileListenerCallback, TrustedPromise),
+    Empty(FileListenerCallback, FileListenerTarget),
+    Receiving(Vec<u8>, FileListenerCallback, FileListenerTarget),
 }
 
 #[derive(JSTraceable, MallocSizeOf)]
@@ -356,6 +365,14 @@ pub enum BlobState {
     UnManaged,
 }
 
+/// The result of looking-up the data for a Blob,
+/// containing either the in-memory bytes,
+/// or the file-id.
+enum BlobResult {
+    Bytes(Vec<u8>),
+    File(Uuid, usize),
+}
+
 /// Data representing a message-port managed by this global.
 #[derive(JSTraceable, MallocSizeOf)]
 #[unrooted_must_root_lint::must_root]
@@ -532,57 +549,137 @@ impl MessageListener {
     }
 }
 
+/// Callback used to enqueue file chunks to streams as part of FileListener.
+fn stream_handle_incoming(stream: &ReadableStream, bytes: Result<Vec<u8>, Error>) {
+    match bytes {
+        Ok(b) => {
+            stream.enqueue_native(b);
+        },
+        Err(e) => {
+            stream.error_native(e);
+        },
+    }
+}
+
+/// Callback used to close streams as part of FileListener.
+fn stream_handle_eof(stream: &ReadableStream) {
+    stream.close_native();
+}
+
 impl FileListener {
     fn handle(&mut self, msg: FileManagerResult<ReadFileProgress>) {
         match msg {
             Ok(ReadFileProgress::Meta(blob_buf)) => match self.state.take() {
-                Some(FileListenerState::Empty(callback, promise)) => {
-                    self.state = Some(FileListenerState::Receiving(
-                        blob_buf.bytes,
-                        callback,
-                        promise,
-                    ));
+                Some(FileListenerState::Empty(callback, target)) => {
+                    let bytes = if let FileListenerTarget::Stream(ref trusted_stream) = target {
+                        let trusted = trusted_stream.clone();
+
+                        let task = task!(enqueue_stream_chunk: move || {
+                            let stream = trusted.root();
+                            stream_handle_incoming(&*stream, Ok(blob_buf.bytes));
+                        });
+
+                        let _ = self
+                            .task_source
+                            .queue_with_canceller(task, &self.task_canceller);
+                        Vec::with_capacity(0)
+                    } else {
+                        blob_buf.bytes
+                    };
+
+                    self.state = Some(FileListenerState::Receiving(bytes, callback, target));
                 },
                 _ => panic!(
                     "Unexpected FileListenerState when receiving ReadFileProgress::Meta msg."
                 ),
             },
             Ok(ReadFileProgress::Partial(mut bytes_in)) => match self.state.take() {
-                Some(FileListenerState::Receiving(mut bytes, callback, promise)) => {
-                    bytes.append(&mut bytes_in);
-                    self.state = Some(FileListenerState::Receiving(bytes, callback, promise));
+                Some(FileListenerState::Receiving(mut bytes, callback, target)) => {
+                    if let FileListenerTarget::Stream(ref trusted_stream) = target {
+                        let trusted = trusted_stream.clone();
+
+                        let task = task!(enqueue_stream_chunk: move || {
+                            let stream = trusted.root();
+                            stream_handle_incoming(&*stream, Ok(bytes_in));
+                        });
+
+                        let _ = self
+                            .task_source
+                            .queue_with_canceller(task, &self.task_canceller);
+                    } else {
+                        bytes.append(&mut bytes_in);
+                    };
+
+                    self.state = Some(FileListenerState::Receiving(bytes, callback, target));
                 },
                 _ => panic!(
                     "Unexpected FileListenerState when receiving ReadFileProgress::Partial msg."
                 ),
             },
             Ok(ReadFileProgress::EOF) => match self.state.take() {
-                Some(FileListenerState::Receiving(bytes, callback, trusted_promise)) => {
-                    let _ = self.task_source.queue_with_canceller(
-                        task!(resolve_promise: move || {
+                Some(FileListenerState::Receiving(bytes, callback, target)) => match target {
+                    FileListenerTarget::Promise(trusted_promise) => {
+                        let callback = match callback {
+                            FileListenerCallback::Promise(callback) => callback,
+                            _ => panic!("Expected promise callback."),
+                        };
+                        let task = task!(resolve_promise: move || {
                             let promise = trusted_promise.root();
                             let _ac = enter_realm(&*promise.global());
-                            callback.0(promise, Ok(bytes));
-                        }),
-                        &self.task_canceller,
-                    );
+                            callback(promise, Ok(bytes));
+                        });
+
+                        let _ = self
+                            .task_source
+                            .queue_with_canceller(task, &self.task_canceller);
+                    },
+                    FileListenerTarget::Stream(trusted_stream) => {
+                        let trusted = trusted_stream.clone();
+
+                        let task = task!(enqueue_stream_chunk: move || {
+                            let stream = trusted.root();
+                            stream_handle_eof(&*stream);
+                        });
+
+                        let _ = self
+                            .task_source
+                            .queue_with_canceller(task, &self.task_canceller);
+                    },
                 },
                 _ => {
                     panic!("Unexpected FileListenerState when receiving ReadFileProgress::EOF msg.")
                 },
             },
             Err(_) => match self.state.take() {
-                Some(FileListenerState::Receiving(_, callback, trusted_promise)) |
-                Some(FileListenerState::Empty(callback, trusted_promise)) => {
-                    let bytes = Err(Error::Network);
-                    let _ = self.task_source.queue_with_canceller(
-                        task!(reject_promise: move || {
-                            let promise = trusted_promise.root();
-                            let _ac = enter_realm(&*promise.global());
-                            callback.0(promise, bytes);
-                        }),
-                        &self.task_canceller,
-                    );
+                Some(FileListenerState::Receiving(_, callback, target)) |
+                Some(FileListenerState::Empty(callback, target)) => {
+                    let error = Err(Error::Network);
+
+                    match target {
+                        FileListenerTarget::Promise(trusted_promise) => {
+                            let callback = match callback {
+                                FileListenerCallback::Promise(callback) => callback,
+                                _ => panic!("Expected promise callback."),
+                            };
+                            let _ = self.task_source.queue_with_canceller(
+                                task!(reject_promise: move || {
+                                    let promise = trusted_promise.root();
+                                    let _ac = enter_realm(&*promise.global());
+                                    callback(promise, error);
+                                }),
+                                &self.task_canceller,
+                            );
+                        },
+                        FileListenerTarget::Stream(trusted_stream) => {
+                            let _ = self.task_source.queue_with_canceller(
+                                task!(error_stream: move || {
+                                    let stream = trusted_stream.root();
+                                    stream_handle_incoming(&*stream, error);
+                                }),
+                                &self.task_canceller,
+                            );
+                        },
+                    }
                 },
                 _ => panic!("Unexpected FileListenerState when receiving Err msg."),
             },
@@ -1565,6 +1662,70 @@ impl GlobalScope {
         }
     }
 
+    /// Get a slice to the inner data of a Blob,
+    /// if it's a memory blob, or it's file-id and file-size otherwise.
+    ///
+    /// Note: this is almost a duplicate of `get_blob_bytes`,
+    /// tweaked for integration with streams.
+    /// TODO: merge with `get_blob_bytes` by way of broader integration with blob streams.
+    fn get_blob_bytes_or_file_id(&self, blob_id: &BlobId) -> BlobResult {
+        let parent = {
+            let blob_state = self.blob_state.borrow();
+            if let BlobState::Managed(blobs_map) = &*blob_state {
+                let blob_info = blobs_map
+                    .get(blob_id)
+                    .expect("get_blob_bytes_or_file_id for an unknown blob.");
+                match blob_info.blob_impl.blob_data() {
+                    BlobData::Sliced(ref parent, ref rel_pos) => {
+                        Some((parent.clone(), rel_pos.clone()))
+                    },
+                    _ => None,
+                }
+            } else {
+                panic!("get_blob_bytes_or_file_id called on a global not managing any blobs.");
+            }
+        };
+
+        match parent {
+            Some((parent_id, rel_pos)) => {
+                match self.get_blob_bytes_non_sliced_or_file_id(&parent_id) {
+                    BlobResult::Bytes(bytes) => {
+                        let range = rel_pos.to_abs_range(bytes.len());
+                        BlobResult::Bytes(bytes.index(range).to_vec())
+                    },
+                    res => res,
+                }
+            },
+            None => self.get_blob_bytes_non_sliced_or_file_id(blob_id),
+        }
+    }
+
+    /// Get bytes from a non-sliced blob if in memory, or it's file-id and file-size.
+    ///
+    /// Note: this is almost a duplicate of `get_blob_bytes_non_sliced`,
+    /// tweaked for integration with streams.
+    /// TODO: merge with `get_blob_bytes` by way of broader integration with blob streams.
+    fn get_blob_bytes_non_sliced_or_file_id(&self, blob_id: &BlobId) -> BlobResult {
+        let blob_state = self.blob_state.borrow();
+        if let BlobState::Managed(blobs_map) = &*blob_state {
+            let blob_info = blobs_map
+                .get(blob_id)
+                .expect("get_blob_bytes_non_sliced_or_file_id called for a unknown blob.");
+            match blob_info.blob_impl.blob_data() {
+                BlobData::File(ref f) => match f.get_cache() {
+                    Some(bytes) => BlobResult::Bytes(bytes.clone()),
+                    None => BlobResult::File(f.get_id(), f.get_size() as usize),
+                },
+                BlobData::Memory(ref s) => BlobResult::Bytes(s.clone()),
+                BlobData::Sliced(_, _) => panic!("This blob doesn't have a parent."),
+            }
+        } else {
+            panic!(
+                "get_blob_bytes_non_sliced_or_file_id called on a global not managing any blobs."
+            );
+        }
+    }
+
     /// Get a copy of the type_string of a blob.
     pub fn get_blob_type_string(&self, blob_id: &BlobId) -> String {
         let blob_state = self.blob_state.borrow();
@@ -1769,6 +1930,50 @@ impl GlobalScope {
         GlobalScope::read_msg(recv)
     }
 
+    /// <https://w3c.github.io/FileAPI/#blob-get-stream>
+    pub fn get_blob_stream(&self, blob_id: &BlobId) -> DomRoot<ReadableStream> {
+        let (file_id, size) = match self.get_blob_bytes_or_file_id(blob_id) {
+            BlobResult::Bytes(bytes) => {
+                // If we have all the bytes in memory, queue them and close the stream.
+                let stream = ReadableStream::new_from_bytes(self, bytes);
+                return stream;
+            },
+            BlobResult::File(id, size) => (id, size),
+        };
+
+        let stream = ReadableStream::new_with_external_underlying_source(
+            self,
+            ExternalUnderlyingSource::Blob(size as usize),
+        );
+
+        let recv = self.send_msg(file_id);
+
+        let trusted_stream = Trusted::new(&*stream.clone());
+        let task_canceller = self.task_canceller(TaskSourceName::FileReading);
+        let task_source = self.file_reading_task_source();
+
+        let mut file_listener = FileListener {
+            state: Some(FileListenerState::Empty(
+                FileListenerCallback::Stream,
+                FileListenerTarget::Stream(trusted_stream),
+            )),
+            task_source,
+            task_canceller,
+        };
+
+        ROUTER.add_route(
+            recv.to_opaque(),
+            Box::new(move |msg| {
+                file_listener.handle(
+                    msg.to()
+                        .expect("Deserialization of file listener msg failed."),
+                );
+            }),
+        );
+
+        stream
+    }
+
     pub fn read_file_async(
         &self,
         id: Uuid,
@@ -1783,8 +1988,8 @@ impl GlobalScope {
 
         let mut file_listener = FileListener {
             state: Some(FileListenerState::Empty(
-                FileListenerCallback(callback),
-                trusted_promise,
+                FileListenerCallback::Promise(callback),
+                FileListenerTarget::Promise(trusted_promise),
             )),
             task_source,
             task_canceller,
@@ -1894,6 +2099,12 @@ impl GlobalScope {
         global_scope_from_global(global, cx)
     }
 
+    /// Returns the global scope for the given SafeJSContext
+    #[allow(unsafe_code)]
+    pub fn from_safe_context(cx: SafeJSContext, realm: InRealm) -> DomRoot<Self> {
+        unsafe { Self::from_context(*cx, realm) }
+    }
+
     /// Returns the global object of the realm that the given JS object
     /// was created in, after unwrapping any wrappers.
     #[allow(unsafe_code)]