diff options
Diffstat (limited to 'runtime')
| -rw-r--r-- | runtime/ops/web_worker/sync_fetch.rs | 202 | ||||
| -rw-r--r-- | runtime/web_worker.rs | 55 |
2 files changed, 132 insertions, 125 deletions
diff --git a/runtime/ops/web_worker/sync_fetch.rs b/runtime/ops/web_worker/sync_fetch.rs index 6ad6edba7..0659241f1 100644 --- a/runtime/ops/web_worker/sync_fetch.rs +++ b/runtime/ops/web_worker/sync_fetch.rs @@ -38,118 +38,124 @@ pub fn op_worker_sync_fetch( let handle = state.borrow::<WebWorkerInternalHandle>().clone(); assert_eq!(handle.worker_type, WebWorkerType::Classic); - // TODO(andreubotella) Make the runtime into a resource and add a new op to - // block on each request, so a script can run while the next loads. - - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_io() - .enable_time() - .build() - .unwrap(); - // TODO(andreubotella) It's not good to throw an exception related to blob // URLs when none of the script URLs use the blob scheme. // Also, in which contexts are blob URLs not supported? - let blob_store = state.try_borrow::<BlobStore>().ok_or_else(|| { - type_error("Blob URLs are not supported in this context.") - })?; - - let handles: Vec<_> = scripts - .into_iter() - .map(|script| -> JoinHandle<Result<SyncFetchScript, AnyError>> { - let blob_store = blob_store.clone(); - runtime.spawn(async move { - let script_url = - Url::parse(&script).map_err(|_| type_error("Invalid script URL"))?; - - let (body, mime_type, res_url) = match script_url.scheme() { - "http" | "https" => { - let resp = reqwest::get(script_url).await?.error_for_status()?; - - let res_url = resp.url().to_string(); - - // TODO(andreubotella) Properly run fetch's "extract a MIME type". - let mime_type = resp - .headers() - .get("Content-Type") - .and_then(|v| v.to_str().ok()) - .map(mime_type_essence); - - // Always check the MIME type with HTTP(S). - loose_mime_checks = false; - - let body = resp.bytes().await?; - - (body, mime_type, res_url) - } - "data" => { - let data_url = DataUrl::process(&script) - .map_err(|e| type_error(format!("{:?}", e)))?; + let blob_store = state + .try_borrow::<BlobStore>() + .ok_or_else(|| type_error("Blob URLs are not supported in this context."))? + .clone(); + + // TODO(andreubotella): make the below thread into a resource that can be + // re-used. This would allow parallel fecthing of multiple scripts. + + let thread = std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_io() + .enable_time() + .build() + .unwrap(); + + let handles: Vec<_> = scripts + .into_iter() + .map(|script| -> JoinHandle<Result<SyncFetchScript, AnyError>> { + let blob_store = blob_store.clone(); + runtime.spawn(async move { + let script_url = Url::parse(&script) + .map_err(|_| type_error("Invalid script URL"))?; + + let (body, mime_type, res_url) = match script_url.scheme() { + "http" | "https" => { + let resp = reqwest::get(script_url).await?.error_for_status()?; + + let res_url = resp.url().to_string(); + + // TODO(andreubotella) Properly run fetch's "extract a MIME type". + let mime_type = resp + .headers() + .get("Content-Type") + .and_then(|v| v.to_str().ok()) + .map(mime_type_essence); + + // Always check the MIME type with HTTP(S). + loose_mime_checks = false; + + let body = resp.bytes().await?; + + (body, mime_type, res_url) + } + "data" => { + let data_url = DataUrl::process(&script) + .map_err(|e| type_error(format!("{:?}", e)))?; - let mime_type = { - let mime = data_url.mime_type(); - format!("{}/{}", mime.type_, mime.subtype) - }; + let mime_type = { + let mime = data_url.mime_type(); + format!("{}/{}", mime.type_, mime.subtype) + }; - let (body, _) = data_url - .decode_to_vec() - .map_err(|e| type_error(format!("{:?}", e)))?; + let (body, _) = data_url + .decode_to_vec() + .map_err(|e| type_error(format!("{:?}", e)))?; - (Bytes::from(body), Some(mime_type), script) - } - "blob" => { - let blob = blob_store - .get_object_url(script_url)? - .ok_or_else(|| type_error("Blob for the given URL not found."))?; + (Bytes::from(body), Some(mime_type), script) + } + "blob" => { + let blob = + blob_store.get_object_url(script_url)?.ok_or_else(|| { + type_error("Blob for the given URL not found.") + })?; - let mime_type = mime_type_essence(&blob.media_type); + let mime_type = mime_type_essence(&blob.media_type); - let body = blob.read_all().await?; + let body = blob.read_all().await?; - (Bytes::from(body), Some(mime_type), script) - } - _ => { - return Err(type_error(format!( - "Classic scripts with scheme {}: are not supported in workers.", - script_url.scheme() - ))) - } - }; - - if !loose_mime_checks { - // TODO(andreubotella) Check properly for a Javascript MIME type. - match mime_type.as_deref() { - Some("application/javascript" | "text/javascript") => {} - Some(mime_type) => { - return Err( - DomExceptionNetworkError { - msg: format!("Invalid MIME type {:?}.", mime_type), - } - .into(), - ) + (Bytes::from(body), Some(mime_type), script) + } + _ => { + return Err(type_error(format!( + "Classic scripts with scheme {}: are not supported in workers.", + script_url.scheme() + ))) } - None => { - return Err( - DomExceptionNetworkError::new("Missing MIME type.").into(), - ) + }; + + if !loose_mime_checks { + // TODO(andreubotella) Check properly for a Javascript MIME type. + match mime_type.as_deref() { + Some("application/javascript" | "text/javascript") => {} + Some(mime_type) => { + return Err( + DomExceptionNetworkError { + msg: format!("Invalid MIME type {:?}.", mime_type), + } + .into(), + ) + } + None => { + return Err( + DomExceptionNetworkError::new("Missing MIME type.").into(), + ) + } } } - } - let (text, _) = encoding_rs::UTF_8.decode_with_bom_removal(&body); + let (text, _) = encoding_rs::UTF_8.decode_with_bom_removal(&body); - Ok(SyncFetchScript { - url: res_url, - script: text.into_owned(), + Ok(SyncFetchScript { + url: res_url, + script: text.into_owned(), + }) }) }) - }) - .collect(); - - let mut ret = Vec::with_capacity(handles.len()); - for handle in handles { - let script = runtime.block_on(handle)??; - ret.push(script); - } - Ok(ret) + .collect(); + + let mut ret = Vec::with_capacity(handles.len()); + for handle in handles { + let script = runtime.block_on(handle)??; + ret.push(script); + } + Ok(ret) + }); + + thread.join().unwrap() } diff --git a/runtime/web_worker.rs b/runtime/web_worker.rs index 240d79d1f..2594b9049 100644 --- a/runtime/web_worker.rs +++ b/runtime/web_worker.rs @@ -11,7 +11,6 @@ use deno_core::error::AnyError; use deno_core::error::JsError; use deno_core::futures::channel::mpsc; use deno_core::futures::future::poll_fn; -use deno_core::futures::future::FutureExt; use deno_core::futures::stream::StreamExt; use deno_core::located_script_name; use deno_core::serde::Deserialize; @@ -568,36 +567,38 @@ pub fn run_web_worker( // TODO(bartlomieju): run following block using "select!" // with terminate - // Execute provided source code immediately - let result = if let Some(source_code) = maybe_source_code { - worker.execute_script(&located_script_name!(), &source_code) - } else { - // TODO(bartlomieju): add "type": "classic", ie. ability to load - // script instead of module - let load_future = worker.execute_module(&specifier).boxed_local(); + let fut = async move { + // Execute provided source code immediately + let result = if let Some(source_code) = maybe_source_code { + worker.execute_script(&located_script_name!(), &source_code) + } else { + // TODO(bartlomieju): add "type": "classic", ie. ability to load + // script instead of module + worker.execute_module(&specifier).await + }; - rt.block_on(load_future) - }; + let internal_handle = worker.internal_handle.clone(); - let internal_handle = worker.internal_handle.clone(); + // If sender is closed it means that worker has already been closed from + // within using "globalThis.close()" + if internal_handle.is_terminated() { + return Ok(()); + } - // If sender is closed it means that worker has already been closed from - // within using "globalThis.close()" - if internal_handle.is_terminated() { - return Ok(()); - } + if let Err(e) = result { + print_worker_error(e.to_string(), &name); + internal_handle + .post_event(WorkerControlEvent::TerminalError(e)) + .expect("Failed to post message to host"); - if let Err(e) = result { - print_worker_error(e.to_string(), &name); - internal_handle - .post_event(WorkerControlEvent::TerminalError(e)) - .expect("Failed to post message to host"); + // Failure to execute script is a terminal error, bye, bye. + return Ok(()); + } - // Failure to execute script is a terminal error, bye, bye. - return Ok(()); - } + let result = worker.run_event_loop(true).await; + debug!("Worker thread shuts down {}", &name); + result + }; - let result = rt.block_on(worker.run_event_loop(true)); - debug!("Worker thread shuts down {}", &name); - result + rt.block_on(fut) } |